struct PriorSet {
	const bool ec_pool;
	set<pg_shard_t> probe; /// current+prior OSDs we need to probe.
	set<int> down;  /// down osds that would normally be in @a probe and might be interesting.
	map<int, epoch_t> blocked_by;  /// current lost_at values for any OSDs in cur set for which (re)marking them lost would affect cur set
	
	bool pg_down;   /// some down osds are included in @a cur; the DOWN pg state bit should be set.
	boost::scoped_ptr<IsPGRecoverablePredicate> pcontdec;
};

/*---------------------------------------------------*/
#undef dout_prefix
#define dout_prefix (*_dout << (debug_pg ? debug_pg->gen_prefix() : string()) << " PriorSet: ")

PG::PriorSet::PriorSet(bool ec_pool,
		       IsPGRecoverablePredicate *c,
		       const OSDMap &osdmap,
		       const map<epoch_t, pg_interval_t> &past_intervals,
		       const vector<int> &up,
		       const vector<int> &acting,
		       const pg_info_t &info,
		       const PG *debug_pg)
  : ec_pool(ec_pool), pg_down(false), pcontdec(c)
{
  /*
   * We have to be careful to gracefully deal with situations like
   * so. Say we have a power outage or something that takes out both
   * OSDs, but the monitor doesn't mark them down in the same epoch.
   * The history may look like
   *
   *  1: A B
   *  2:   B
   *  3:       let's say B dies for good, too (say, from the power spike) 
   *  4: A
   *
   * which makes it look like B may have applied updates to the PG
   * that we need in order to proceed.  This sucks...
   *
   * To minimize the risk of this happening, we CANNOT go active if
   * _any_ OSDs in the prior set are down until we send an MOSDAlive
   * to the monitor such that the OSDMap sets osd_up_thru to an epoch.
   * Then, we have something like
   *
   *  1: A B
   *  2:   B   up_thru[B]=0
   *  3:
   *  4: A
   *
   * -> we can ignore B, bc it couldn't have gone active (alive_thru
   *    still 0).
   *
   * or,
   *
   *  1: A B
   *  2:   B   up_thru[B]=0
   *  3:   B   up_thru[B]=2
   *  4:
   *  5: A    
   *
   * -> we must wait for B, bc it was alive through 2, and could have
   *    written to the pg.
   *
   * If B is really dead, then an administrator will need to manually
   * intervene by marking the OSD as "lost."
   */

	// Include current acting and up nodes... not because they may
	// contain old data (this interval hasn't gone active, obviously),
	// but because we want their pg_info to inform choose_acting(), and
	// so that we know what they do/do not have explicitly before
	// sending them any new info/logs/whatever.
	for (unsigned i=0; i<acting.size(); i++) {
		if (acting[i] != CRUSH_ITEM_NONE)
			probe.insert(pg_shard_t(acting[i], ec_pool ? shard_id_t(i) : shard_id_t::NO_SHARD));
	}

	// It may be possible to exlude the up nodes, but let's keep them in
	// there for now.
	for (unsigned i=0; i<up.size(); i++) {
		if (up[i] != CRUSH_ITEM_NONE)
			probe.insert(pg_shard_t(up[i], ec_pool ? shard_id_t(i) : shard_id_t::NO_SHARD));
	}
	
	for (map<epoch_t,pg_interval_t>::const_reverse_iterator p = past_intervals.rbegin();p != past_intervals.rend();++p) {
		const pg_interval_t &interval = p->second;
		dout(10) << "build_prior " << interval << dendl;
	
		if (interval.last < info.history.last_epoch_started)
			break;  // we don't care
	
		if (interval.acting.empty())
			continue;
	
		if (!interval.maybe_went_rw)
			continue;

	    // look at candidate osds during this interval.  each falls into
	    // one of three categories: up, down (but potentially
	    // interesting), or lost (down, but we won't wait for it).
    	set<pg_shard_t> up_now;
		bool any_down_now = false;  // any candidates down now (that might have useful data)

		// consider ACTING osds
		for (unsigned i=0; i<interval.acting.size(); i++) {
			int o = interval.acting[i];
			if (o == CRUSH_ITEM_NONE)
				continue;

			pg_shard_t so(o, ec_pool ? shard_id_t(i) : shard_id_t::NO_SHARD);
		
			const osd_info_t *pinfo = 0;
			if (osdmap.exists(o))
				pinfo = &osdmap.get_info(o);
		
			if (osdmap.is_up(o)) {
				// include past acting osds if they are up.
				probe.insert(so);
				up_now.insert(so);
			} else if (!pinfo) {
				dout(10) << "build_prior  prior osd." << o << " no longer exists" << dendl;
				down.insert(o);
			} else if (pinfo->lost_at > interval.first) {
				dout(10) << "build_prior  prior osd." << o << " is down, but lost_at " << pinfo->lost_at << dendl;
				up_now.insert(so);
				down.insert(o);
			} else {
				dout(10) << "build_prior  prior osd." << o << " is down" << dendl;
				down.insert(o);
				any_down_now = true;
			}
		}
		
		// if not enough osds survived this interval, and we may have gone rw,
		// then we need to wait for one of those osds to recover to
		// ensure that we haven't lost any information.
		if (!(*pcontdec)(up_now) && any_down_now) {
			// fixme: how do we identify a "clean" shutdown anyway?

			dout(10) << "build_prior  possibly went active+rw, insufficient up;"<< " including down osds" << dendl;
			for (vector<int>::const_iterator i = interval.acting.begin();i != interval.acting.end();++i) {
				if (osdmap.exists(*i) &&   // if it doesn't exist, we already consider it lost.
				  osdmap.is_down(*i)) {
					pg_down = true;
			
					// make note of when any down osd in the cur set was lost, so that
					// we can notice changes in prior_set_affected.
					blocked_by[*i] = osdmap.get_info(*i).lost_at;
				}
			}
		}
	}

  dout(10) << "build_prior final: probe " << probe<< " down " << down<< " blocked_by " << blocked_by<< (pg_down ? " pg_down":"")<< dendl;
}

void PG::build_prior(std::unique_ptr<PriorSet> &prior_set)
{
	if (1) {
		// sanity check
		for (map<pg_shard_t,pg_info_t>::iterator it = peer_info.begin();it != peer_info.end();++it) {
			assert(info.history.last_epoch_started >= it->second.history.last_epoch_started);
		}
	}

	prior_set.reset(
	  new PriorSet(
		pool.info.ec_pool(),
		get_pgbackend()->get_is_recoverable_predicate(),
		*get_osdmap(),
		past_intervals,
		up,
		acting,
		info,
		this));
	PriorSet &prior(*prior_set.get());
	
	if (prior.pg_down) {
		state_set(PG_STATE_DOWN);
	}

	if (get_osdmap()->get_up_thru(osd->whoami) < info.history.same_interval_since) {
		dout(10) << "up_thru " << get_osdmap()->get_up_thru(osd->whoami)<< " < same_since " << info.history.same_interval_since<< ", must notify monitor" << dendl;
		need_up_thru = true;
	} else {
		dout(10) << "up_thru " << get_osdmap()->get_up_thru(osd->whoami)<< " >= same_since " << info.history.same_interval_since<< ", all is well" << dendl;
		need_up_thru = false;
	}
	set_probe_targets(prior_set->probe);
}

void PG::set_probe_targets(const set<pg_shard_t> &probe_set)
{
	Mutex::Locker l(heartbeat_peer_lock);
	probe_targets.clear();
	for (set<pg_shard_t>::iterator i = probe_set.begin();i != probe_set.end();++i) {
		probe_targets.insert(i->osd);
	}
}

// true if the given map affects the prior set
bool PG::PriorSet::affected_by_map(const OSDMapRef osdmap, const PG *debug_pg) const
{
	for (set<pg_shard_t>::iterator p = probe.begin();p != probe.end();++p) {
		int o = p->osd;
	
		// did someone in the prior set go down?
		if (osdmap->is_down(o) && down.count(o) == 0) {
			dout(10) << "affected_by_map osd." << o << " now down" << dendl;
			return true;
		}
	
		// did a down osd in cur get (re)marked as lost?
		map<int, epoch_t>::const_iterator r = blocked_by.find(o);
		if (r != blocked_by.end()) {
			if (!osdmap->exists(o)) {
				dout(10) << "affected_by_map osd." << o << " no longer exists" << dendl;
				return true;
			}
			if (osdmap->get_info(o).lost_at != r->second) {
				dout(10) << "affected_by_map osd." << o << " (re)marked as lost" << dendl;
				return true;
			}
		}
	}

	// did someone in the prior down set go up?
	for (set<int>::const_iterator p = down.begin();p != down.end();++p) {
		int o = *p;
	
		if (osdmap->is_up(o)) {
			dout(10) << "affected_by_map osd." << *p << " now up" << dendl;
			return true;
		}
	
		// did someone in the prior set get lost or destroyed?
		if (!osdmap->exists(o)) {
			dout(10) << "affected_by_map osd." << o << " no longer exists" << dendl;
			return true;
		}

		// did a down osd in down get (re)marked as lost?
		map<int, epoch_t>::const_iterator r = blocked_by.find(o);
		if (r != blocked_by.end()) {
			if (osdmap->get_info(o).lost_at != r->second) {
				dout(10) << "affected_by_map osd." << o << " (re)marked as lost" << dendl;
				return true;
			}
		} 
	}

  return false;
}

/**
 * pg_info_t - summary of PG statistics.
 *
 * some notes: 
 *  - last_complete implies we have all objects that existed as of that
 *    stamp, OR a newer object, OR have already applied a later delete.
 *  - if last_complete >= log.bottom, then we know pg contents thru log.head.
 *    otherwise, we have no idea what the pg is supposed to contain.
 */
struct pg_info_t {
	spg_t pgid;
	eversion_t last_update;      ///< last object version applied to store.
	eversion_t last_complete;    ///< last version pg was complete through.
	epoch_t last_epoch_started;  ///< last epoch at which this pg started on this osd
	
	version_t last_user_version; ///< last user object version applied to store
	
	eversion_t log_tail;         ///< oldest log entry.
	
	hobject_t last_backfill;     ///< objects >= this and < last_complete may be missing
	bool last_backfill_bitwise;  ///< true if last_backfill reflects a bitwise (vs nibblewise) sort
	
	interval_set<snapid_t> purged_snaps;
	
	pg_stat_t stats;
	
	pg_history_t history;
	pg_hit_set_history_t hit_set;
};

/**
* pg_history_t - information about recent pg peering/mapping history
*
* This is aggressively shared between OSDs to bound the amount of past
* history they need to worry about.
*/
struct pg_history_t {
	epoch_t epoch_created;                  // epoch in which PG was created
	epoch_t last_epoch_started;             // lower bound on last epoch started (anywhere, not necessarily locally)
	epoch_t last_epoch_clean;               // lower bound on last epoch the PG was completely clean.
	epoch_t last_epoch_split;               // as parent
	epoch_t last_epoch_marked_full;         // pool or cluster
	
	/**
	* In the event of a map discontinuity, same_*_since may reflect the first
	* map the osd has seen in the new map sequence rather than the actual start
	* of the interval.  This is ok since a discontinuity at epoch e means there
	* must have been a clean interval between e and now and that we cannot be
	* in the active set during the interval containing e.
	*/
	epoch_t same_up_since;                  // same acting set since
	epoch_t same_interval_since;            // same acting AND up set since
	epoch_t same_primary_since;             // same primary at least back through this epoch.
	
	eversion_t last_scrub;
	eversion_t last_deep_scrub;
	utime_t last_scrub_stamp;
	utime_t last_deep_scrub_stamp;
	utime_t last_clean_scrub_stamp;
};

int OSD::init()
{
	...
	consume_map();
	...
}

void OSD::consume_map()
{
	...
}

bool OSD::ms_dispatch(Message *m)
{
	....

	while (dispatch_running) {
	dout(10) << "ms_dispatch waiting for other dispatch thread to complete" << dendl;
		dispatch_cond.Wait(osd_lock);
	}
	dispatch_running = true;
	
	do_waiters();
	_dispatch(m);
	do_waiters();
	
	dispatch_running = false;
	dispatch_cond.Signal();
	...
}

void OSD::do_waiters()
{
	assert(osd_lock.is_locked());
	
	dout(10) << "do_waiters -- start" << dendl;
	finished_lock.Lock();
	while (!finished.empty()) {
		OpRequestRef next = finished.front();
		finished.pop_front();
		finished_lock.Unlock();
		dispatch_op(next);
		finished_lock.Lock();
	}
	finished_lock.Unlock();
	dout(10) << "do_waiters -- finish" << dendl;
}

void OSD::activate_map(){
	...
	// process waiters
	take_waiters(waiting_for_osdmap);
}

void take_waiters(list<OpRequestRef>& ls) {
	finished_lock.Lock();
	finished.splice(finished.end(), ls);
	finished_lock.Unlock();
}

void OSD::_dispatch(Message *m)
{
	...
	switch (m->get_type()) {
		// map and replication
		case CEPH_MSG_OSD_MAP:
		handle_osd_map(static_cast<MOSDMap*>(m));
		break;
	}
}

void OSD::_committed_osd_maps(epoch_t first, epoch_t last, MOSDMap *m)
{
	...

	// advance through the new maps
	for (epoch_t cur = first; cur <= last; cur++) {
		
		OSDMapRef newmap = get_map(cur);
		service.pre_publish_map(newmap);               //将新的osdmap标记为预发布状态

		/*
		 * kill connections to newly down osds
		 * （等待当前OSDMap上的请求处理完成）
		 */
		bool waited_for_reservations = false;
		set<int> old;
		osdmap->get_all_osds(old);
		for (set<int>::iterator p = old.begin(); p != old.end(); ++p) {
			if (*p != whoami &&
			  osdmap->have_inst(*p) &&                        // in old map
			  (!newmap->exists(*p) || !newmap->is_up(*p))) {  // but not the new one
				if (!waited_for_reservations) {
					service.await_reserved_maps();
					waited_for_reservations = true;
				}
				note_down_osd(*p);
			}
		}

		//将newmap发布
		osdmap = newmap;
	}
}

void OSD::_committed_osd_maps(epoch_t first, epoch_t last, MOSDMap *m)
{	
	...

	if (osdmap->get_epoch() > 0 && is_active()) {
		if (!osdmap->exists(whoami)) {

			//当前OSD已经在OSDMap中不存在了，则发起相应的信号关闭当前OSD

	}else if (!osdmap->is_up(whoami) ||
		!osdmap->get_addr(whoami).probably_equals(
		client_messenger->get_myaddr()) ||
		!osdmap->get_cluster_addr(whoami).probably_equals(
		cluster_messenger->get_myaddr()) ||
		!osdmap->get_hb_back_addr(whoami).probably_equals(
		hb_back_server_messenger->get_myaddr()) ||
		(osdmap->get_hb_front_addr(whoami) != entity_addr_t() &&
		!osdmap->get_hb_front_addr(whoami).probably_equals(
		hb_front_server_messenger->get_myaddr()))){
			
			//当前OSD的绑定信息出现了错误，需要重新绑定


		}
	}
}

void OSD::_committed_osd_maps(epoch_t first, epoch_t last, MOSDMap *m)
{
	...

	// yay!
	consume_map();
	
	if (is_active() || is_waiting_for_healthy())
	maybe_update_heartbeat_peers();
	
	if (!is_active()) {
		dout(10) << " not yet active; waiting for peering wq to drain" << dendl;
		peering_wq.drain();
	} else {
		activate_map();
	}
	...
}

void OSD::activate_map()
{
	...
	
	service.activate_map();
	// process waiters
	take_waiters(waiting_for_osdmap);
}

void take_waiters(list<OpRequestRef>& ls) {
	finished_lock.Lock();
	finished.splice(finished.end(), ls);
	finished_lock.Unlock();
}

void OSD::consume_map()
{
	 // scan pg's
	{
		RWLock::RLocker l(pg_map_lock);
		for (ceph::unordered_map<spg_t,PG*>::iterator it = pg_map.begin();it != pg_map.end();++it) {
			PG *pg = it->second;
	
			pg->lock();
			if (pg->is_primary())
				num_pg_primary++;
			else if (pg->is_replica())
				num_pg_replica++;
			else
				num_pg_stray++;
		
			if (!osdmap->have_pg_pool(pg->info.pgid.pool())) {
				//pool is deleted!
				to_remove.push_back(PGRef(pg));
			} else {
				service.init_splits_between(it->first, service.get_osdmap(), osdmap);
			}
			
				pg->unlock();
		}
	}

	for (list<PGRef>::iterator i = to_remove.begin();i != to_remove.end();to_remove.erase(i++)) {
		RWLock::WLocker locker(pg_map_lock);
		(*i)->lock();
		_remove_pg(&**i);
		(*i)->unlock();
	}
	to_remove.clear();
}

void OSD::consume_map()
{
	...

	service.pre_publish_map(osdmap);
	service.await_reserved_maps();
	service.publish_map(osdmap);
	
	dispatch_sessions_waiting_on_map();
}

void OSD::consume_map()
{
	// remove any PGs which we no longer host from the session waiting_for_pg lists
	set<spg_t> pgs_to_check;
	get_pgs_with_waiting_sessions(&pgs_to_check);
	for (set<spg_t>::iterator p = pgs_to_check.begin();p != pgs_to_check.end();++p) {
		if (!(osdmap->is_acting_osd_shard(p->pgid, whoami, p->shard))) {
			set<Session*> concerned_sessions;
			get_sessions_possibly_interested_in_pg(*p, &concerned_sessions);
			for (set<Session*>::iterator i = concerned_sessions.begin();i != concerned_sessions.end();++i) {
			  {
				Mutex::Locker l((*i)->session_dispatch_lock);
				session_notify_pg_cleared(*i, osdmap, *p);
			  }
			  (*i)->put();
			}
		}
	}
}

void OSD::consume_map()
{
	// scan pg's
	{
		RWLock::RLocker l(pg_map_lock);
		for (ceph::unordered_map<spg_t,PG*>::iterator it = pg_map.begin();it != pg_map.end();++it) {
			PG *pg = it->second;
			pg->lock();
			pg->queue_null(osdmap->get_epoch(), osdmap->get_epoch());
			pg->unlock();
		}
	
		logger->set(l_osd_pg, pg_map.size());
	}
}


void PG::queue_null(epoch_t msg_epoch,epoch_t query_epoch)
{
	dout(10) << "null" << dendl;
	queue_peering_event(
    	CephPeeringEvtRef(std::make_shared<CephPeeringEvt>(msg_epoch, query_epoch,
					 NullEvt())));
}

void PG::queue_peering_event(CephPeeringEvtRef evt)
{
	if (old_peering_evt(evt))
		return;

	peering_queue.push_back(evt);
	osd->queue_for_peering(this);
}

void OSDService::queue_for_peering(PG *pg)
{
	peering_wq.queue(pg);
}