[ceph.git] / ceph / src / osd / ReplicatedBackend.h

// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
// vim: ts=8 sw=2 smarttab
/*
 * Ceph - scalable distributed file system
 *
 * Copyright (C) 2013 Inktank Storage, Inc.
 *
 * This is free software; you can redistribute it and/or
 * modify it under the terms of the GNU Lesser General Public
 * License version 2.1, as published by the Free Software
 * Foundation.  See file COPYING.
 *
 */

#ifndef REPBACKEND_H
#define REPBACKEND_H

#include "PGBackend.h"

struct C_ReplicatedBackend_OnPullComplete;
class ReplicatedBackend : public PGBackend {
  struct RPGHandle : public PGBackend::RecoveryHandle {
    map<pg_shard_t, vector<PushOp> > pushes;
    map<pg_shard_t, vector<PullOp> > pulls;
  };
  friend struct C_ReplicatedBackend_OnPullComplete;
public:
  ReplicatedBackend(
    PGBackend::Listener *pg,
    const coll_t &coll,
    ObjectStore::CollectionHandle &ch,
    ObjectStore *store,
    CephContext *cct);

  /// @see PGBackend::open_recovery_op
  RPGHandle *_open_recovery_op() {
    return new RPGHandle();
  }
  PGBackend::RecoveryHandle *open_recovery_op() override {
    return _open_recovery_op();
  }

  /// @see PGBackend::run_recovery_op
  void run_recovery_op(
    PGBackend::RecoveryHandle *h,
    int priority) override;

  /// @see PGBackend::recover_object
  int recover_object(
    const hobject_t &hoid,
    eversion_t v,
    ObjectContextRef head,
    ObjectContextRef obc,
    RecoveryHandle *h
    ) override;

  void check_recovery_sources(const OSDMapRef& osdmap) override;

  bool can_handle_while_inactive(OpRequestRef op) override;

  /// @see PGBackend::handle_message
  bool _handle_message(
    OpRequestRef op
    ) override;

  void on_change() override;
  void clear_recovery_state() override;

  class RPCRecPred : public IsPGRecoverablePredicate {
  public:
    bool operator()(const set<pg_shard_t> &have) const override {
      return !have.empty();
    }
  };
  IsPGRecoverablePredicate *get_is_recoverable_predicate() const override {
    return new RPCRecPred;
  }

  class RPCReadPred : public IsPGReadablePredicate {
    pg_shard_t whoami;
  public:
    explicit RPCReadPred(pg_shard_t whoami) : whoami(whoami) {}
    bool operator()(const set<pg_shard_t> &have) const override {
      return have.count(whoami);
    }
  };
  IsPGReadablePredicate *get_is_readable_predicate() const override {
    return new RPCReadPred(get_parent()->whoami_shard());
  }

  void dump_recovery_info(Formatter *f) const override {
    {
      f->open_array_section("pull_from_peer");
      for (map<pg_shard_t, set<hobject_t> >::const_iterator i = pull_from_peer.begin();
	   i != pull_from_peer.end();
	   ++i) {
	f->open_object_section("pulling_from");
	f->dump_stream("pull_from") << i->first;
	{
	  f->open_array_section("pulls");
	  for (set<hobject_t>::const_iterator j = i->second.begin();
	       j != i->second.end();
	       ++j) {
	    f->open_object_section("pull_info");
	    ceph_assert(pulling.count(*j));
	    pulling.find(*j)->second.dump(f);
	    f->close_section();
	  }
	  f->close_section();
	}
	f->close_section();
      }
      f->close_section();
    }
    {
      f->open_array_section("pushing");
      for (map<hobject_t, map<pg_shard_t, PushInfo>>::const_iterator i =
	     pushing.begin();
	   i != pushing.end();
	   ++i) {
	f->open_object_section("object");
	f->dump_stream("pushing") << i->first;
	{
	  f->open_array_section("pushing_to");
	  for (map<pg_shard_t, PushInfo>::const_iterator j = i->second.begin();
	       j != i->second.end();
	       ++j) {
	    f->open_object_section("push_progress");
	    f->dump_stream("pushing_to") << j->first;
	    {
	      f->open_object_section("push_info");
	      j->second.dump(f);
	      f->close_section();
	    }
	    f->close_section();
	  }
	  f->close_section();
	}
	f->close_section();
      }
      f->close_section();
    }
  }

  int objects_read_sync(
    const hobject_t &hoid,
    uint64_t off,
    uint64_t len,
    uint32_t op_flags,
    bufferlist *bl) override;

  int objects_readv_sync(
    const hobject_t &hoid,
    map<uint64_t, uint64_t>&& m,
    uint32_t op_flags,
    bufferlist *bl) override;

  void objects_read_async(
    const hobject_t &hoid,
    const list<pair<boost::tuple<uint64_t, uint64_t, uint32_t>,
	       pair<bufferlist*, Context*> > > &to_read,
               Context *on_complete,
               bool fast_read = false) override;

private:
  // push
  struct PushInfo {
    ObjectRecoveryProgress recovery_progress;
    ObjectRecoveryInfo recovery_info;
    ObjectContextRef obc;
    object_stat_sum_t stat;
    ObcLockManager lock_manager;

    void dump(Formatter *f) const {
      {
	f->open_object_section("recovery_progress");
	recovery_progress.dump(f);
	f->close_section();
      }
      {
	f->open_object_section("recovery_info");
	recovery_info.dump(f);
	f->close_section();
      }
    }
  };
  map<hobject_t, map<pg_shard_t, PushInfo>> pushing;

  // pull
  struct PullInfo {
    pg_shard_t from;
    hobject_t soid;
    ObjectRecoveryProgress recovery_progress;
    ObjectRecoveryInfo recovery_info;
    ObjectContextRef head_ctx;
    ObjectContextRef obc;
    object_stat_sum_t stat;
    bool cache_dont_need;
    ObcLockManager lock_manager;

    void dump(Formatter *f) const {
      {
	f->open_object_section("recovery_progress");
	recovery_progress.dump(f);
	f->close_section();
      }
      {
	f->open_object_section("recovery_info");
	recovery_info.dump(f);
	f->close_section();
      }
    }

    bool is_complete() const {
      return recovery_progress.is_complete(recovery_info);
    }
  };

  map<hobject_t, PullInfo> pulling;

  // Reverse mapping from osd peer to objects being pulled from that peer
  map<pg_shard_t, set<hobject_t> > pull_from_peer;
  void clear_pull(
    map<hobject_t, PullInfo>::iterator piter,
    bool clear_pull_from_peer = true);
  void clear_pull_from(
    map<hobject_t, PullInfo>::iterator piter);

  void _do_push(OpRequestRef op);
  void _do_pull_response(OpRequestRef op);
  void do_push(OpRequestRef op) {
    if (is_primary()) {
      _do_pull_response(op);
    } else {
      _do_push(op);
    }
  }
  void do_pull(OpRequestRef op);
  void do_push_reply(OpRequestRef op);

  bool handle_push_reply(pg_shard_t peer, const PushReplyOp &op, PushOp *reply);
  void handle_pull(pg_shard_t peer, PullOp &op, PushOp *reply);

  struct pull_complete_info {
    hobject_t hoid;
    object_stat_sum_t stat;
  };
  bool handle_pull_response(
    pg_shard_t from, const PushOp &op, PullOp *response,
    list<pull_complete_info> *to_continue,
    ObjectStore::Transaction *t);
  void handle_push(pg_shard_t from, const PushOp &op, PushReplyOp *response,
		   ObjectStore::Transaction *t, bool is_repair);

  static void trim_pushed_data(const interval_set<uint64_t> &copy_subset,
			       const interval_set<uint64_t> &intervals_received,
			       bufferlist data_received,
			       interval_set<uint64_t> *intervals_usable,
			       bufferlist *data_usable);
  void _failed_pull(pg_shard_t from, const hobject_t &soid);

  void send_pushes(int prio, map<pg_shard_t, vector<PushOp> > &pushes);
  void prep_push_op_blank(const hobject_t& soid, PushOp *op);
  void send_pulls(
    int priority,
    map<pg_shard_t, vector<PullOp> > &pulls);

  int build_push_op(const ObjectRecoveryInfo &recovery_info,
		    const ObjectRecoveryProgress &progress,
		    ObjectRecoveryProgress *out_progress,
		    PushOp *out_op,
		    object_stat_sum_t *stat = 0,
                    bool cache_dont_need = true);
  void submit_push_data(const ObjectRecoveryInfo &recovery_info,
			bool first,
			bool complete,
			bool clear_omap,
			bool cache_dont_need,
			interval_set<uint64_t> &data_zeros,
			const interval_set<uint64_t> &intervals_included,
			bufferlist data_included,
			bufferlist omap_header,
			const map<string, bufferlist> &attrs,
			const map<string, bufferlist> &omap_entries,
			ObjectStore::Transaction *t);
  void submit_push_complete(const ObjectRecoveryInfo &recovery_info,
			    ObjectStore::Transaction *t);

  void calc_clone_subsets(
    SnapSet& snapset, const hobject_t& poid, const pg_missing_t& missing,
    const hobject_t &last_backfill,
    interval_set<uint64_t>& data_subset,
    map<hobject_t, interval_set<uint64_t>>& clone_subsets,
    ObcLockManager &lock_manager);
  void prepare_pull(
    eversion_t v,
    const hobject_t& soid,
    ObjectContextRef headctx,
    RPGHandle *h);
  int start_pushes(
    const hobject_t &soid,
    ObjectContextRef obj,
    RPGHandle *h);
  int prep_push_to_replica(
    ObjectContextRef obc, const hobject_t& soid, pg_shard_t peer,
    PushOp *pop, bool cache_dont_need = true);
  int prep_push(
    ObjectContextRef obc,
    const hobject_t& oid, pg_shard_t dest,
    PushOp *op,
    bool cache_dont_need);
  int prep_push(
    ObjectContextRef obc,
    const hobject_t& soid, pg_shard_t peer,
    eversion_t version,
    interval_set<uint64_t> &data_subset,
    map<hobject_t, interval_set<uint64_t>>& clone_subsets,
    PushOp *op,
    bool cache,
    ObcLockManager &&lock_manager);
  void calc_head_subsets(
    ObjectContextRef obc, SnapSet& snapset, const hobject_t& head,
    const pg_missing_t& missing,
    const hobject_t &last_backfill,
    interval_set<uint64_t>& data_subset,
    map<hobject_t, interval_set<uint64_t>>& clone_subsets,
    ObcLockManager &lock_manager);
  ObjectRecoveryInfo recalc_subsets(
    const ObjectRecoveryInfo& recovery_info,
    SnapSetContext *ssc,
    ObcLockManager &lock_manager);

  /**
   * Client IO
   */
  struct InProgressOp : public RefCountedObject {
    ceph_tid_t tid;
    set<pg_shard_t> waiting_for_commit;
    Context *on_commit;
    OpRequestRef op;
    eversion_t v;
    bool done() const {
      return waiting_for_commit.empty();
    }
  private:
    FRIEND_MAKE_REF(InProgressOp);
    InProgressOp(ceph_tid_t tid, Context *on_commit, OpRequestRef op, eversion_t v)
      :
	tid(tid), on_commit(on_commit),
	op(op), v(v) {}
  };
  map<ceph_tid_t, ceph::ref_t<InProgressOp>> in_progress_ops;
public:
  friend class C_OSD_OnOpCommit;

  void call_write_ordered(std::function<void(void)> &&cb) override {
    // ReplicatedBackend submits writes inline in submit_transaction, so
    // we can just call the callback.
    cb();
  }

  void submit_transaction(
    const hobject_t &hoid,
    const object_stat_sum_t &delta_stats,
    const eversion_t &at_version,
    PGTransactionUPtr &&t,
    const eversion_t &trim_to,
    const eversion_t &min_last_complete_ondisk,
    const vector<pg_log_entry_t> &log_entries,
    std::optional<pg_hit_set_history_t> &hset_history,
    Context *on_all_commit,
    ceph_tid_t tid,
    osd_reqid_t reqid,
    OpRequestRef op
    ) override;

private:
  Message * generate_subop(
    const hobject_t &soid,
    const eversion_t &at_version,
    ceph_tid_t tid,
    osd_reqid_t reqid,
    eversion_t pg_trim_to,
    eversion_t min_last_complete_ondisk,
    hobject_t new_temp_oid,
    hobject_t discard_temp_oid,
    const bufferlist &log_entries,
    std::optional<pg_hit_set_history_t> &hset_history,
    ObjectStore::Transaction &op_t,
    pg_shard_t peer,
    const pg_info_t &pinfo);
  void issue_op(
    const hobject_t &soid,
    const eversion_t &at_version,
    ceph_tid_t tid,
    osd_reqid_t reqid,
    eversion_t pg_trim_to,
    eversion_t min_last_complete_ondisk,
    hobject_t new_temp_oid,
    hobject_t discard_temp_oid,
    const vector<pg_log_entry_t> &log_entries,
    std::optional<pg_hit_set_history_t> &hset_history,
    InProgressOp *op,
    ObjectStore::Transaction &op_t);
  void op_commit(const ceph::ref_t<InProgressOp>& op);
  void do_repop_reply(OpRequestRef op);
  void do_repop(OpRequestRef op);

  struct RepModify {
    OpRequestRef op;
    bool committed;
    int ackerosd;
    eversion_t last_complete;
    epoch_t epoch_started;

    ObjectStore::Transaction opt, localt;
    
    RepModify() : committed(false), ackerosd(-1),
		  epoch_started(0) {}
  };
  typedef std::shared_ptr<RepModify> RepModifyRef;

  struct C_OSD_RepModifyCommit;

  void repop_commit(RepModifyRef rm);
  bool auto_repair_supported() const override { return store->has_builtin_csum(); }


  int be_deep_scrub(
    const hobject_t &poid,
    ScrubMap &map,
    ScrubMapBuilder &pos,
    ScrubMap::object &o) override;
  uint64_t be_get_ondisk_size(uint64_t logical_size) override { return logical_size; }
};

#endif
Commit	Line	Data
7c673cae FG	1	// -- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t --
	2	// vim: ts=8 sw=2 smarttab
	3	/*
	4	* Ceph - scalable distributed file system
	5	*
	6	* Copyright (C) 2013 Inktank Storage, Inc.
	7	*
	8	* This is free software; you can redistribute it and/or
	9	* modify it under the terms of the GNU Lesser General Public
	10	* License version 2.1, as published by the Free Software
	11	* Foundation. See file COPYING.
	12	*
	13	*/
	14
	15	#ifndef REPBACKEND_H
	16	#define REPBACKEND_H
	17
7c673cae	18	#include "PGBackend.h"
7c673cae FG	19
	20	struct C_ReplicatedBackend_OnPullComplete;
	21	class ReplicatedBackend : public PGBackend {
	22	struct RPGHandle : public PGBackend::RecoveryHandle {
	23	map<pg_shard_t, vector<PushOp> > pushes;
	24	map<pg_shard_t, vector<PullOp> > pulls;
	25	};
	26	friend struct C_ReplicatedBackend_OnPullComplete;
	27	public:
	28	ReplicatedBackend(
	29	PGBackend::Listener *pg,
11fdf7f2	30	const coll_t &coll,
7c673cae FG	31	ObjectStore::CollectionHandle &ch,
	32	ObjectStore *store,
	33	CephContext *cct);
	34
	35	/// @see PGBackend::open_recovery_op
	36	RPGHandle *_open_recovery_op() {
	37	return new RPGHandle();
	38	}
	39	PGBackend::RecoveryHandle *open_recovery_op() override {
	40	return _open_recovery_op();
	41	}
	42
	43	/// @see PGBackend::run_recovery_op
	44	void run_recovery_op(
	45	PGBackend::RecoveryHandle *h,
	46	int priority) override;
	47
	48	/// @see PGBackend::recover_object
224ce89b	49	int recover_object(
7c673cae FG	50	const hobject_t &hoid,
	51	eversion_t v,
	52	ObjectContextRef head,
	53	ObjectContextRef obc,
	54	RecoveryHandle *h
	55	) override;
	56
	57	void check_recovery_sources(const OSDMapRef& osdmap) override;
	58
7c673cae FG	59	bool can_handle_while_inactive(OpRequestRef op) override;
	60
	61	/// @see PGBackend::handle_message
c07f9fc5	62	bool _handle_message(
7c673cae FG	63	OpRequestRef op
	64	) override;
	65
	66	void on_change() override;
	67	void clear_recovery_state() override;
7c673cae FG	68
	69	class RPCRecPred : public IsPGRecoverablePredicate {
	70	public:
	71	bool operator()(const set<pg_shard_t> &have) const override {
	72	return !have.empty();
	73	}
	74	};
11fdf7f2	75	IsPGRecoverablePredicate *get_is_recoverable_predicate() const override {
7c673cae FG	76	return new RPCRecPred;
	77	}
	78
	79	class RPCReadPred : public IsPGReadablePredicate {
	80	pg_shard_t whoami;
	81	public:
	82	explicit RPCReadPred(pg_shard_t whoami) : whoami(whoami) {}
	83	bool operator()(const set<pg_shard_t> &have) const override {
	84	return have.count(whoami);
	85	}
	86	};
11fdf7f2	87	IsPGReadablePredicate *get_is_readable_predicate() const override {
7c673cae FG	88	return new RPCReadPred(get_parent()->whoami_shard());
	89	}
	90
	91	void dump_recovery_info(Formatter *f) const override {
	92	{
	93	f->open_array_section("pull_from_peer");
	94	for (map<pg_shard_t, set<hobject_t> >::const_iterator i = pull_from_peer.begin();
	95	i != pull_from_peer.end();
	96	++i) {
	97	f->open_object_section("pulling_from");
	98	f->dump_stream("pull_from") << i->first;
	99	{
	100	f->open_array_section("pulls");
	101	for (set<hobject_t>::const_iterator j = i->second.begin();
	102	j != i->second.end();
	103	++j) {
	104	f->open_object_section("pull_info");
11fdf7f2	105	ceph_assert(pulling.count(*j));
7c673cae FG	106	pulling.find(*j)->second.dump(f);
	107	f->close_section();
	108	}
	109	f->close_section();
	110	}
	111	f->close_section();
	112	}
	113	f->close_section();
	114	}
	115	{
	116	f->open_array_section("pushing");
	117	for (map<hobject_t, map<pg_shard_t, PushInfo>>::const_iterator i =
	118	pushing.begin();
	119	i != pushing.end();
	120	++i) {
	121	f->open_object_section("object");
	122	f->dump_stream("pushing") << i->first;
	123	{
	124	f->open_array_section("pushing_to");
	125	for (map<pg_shard_t, PushInfo>::const_iterator j = i->second.begin();
	126	j != i->second.end();
	127	++j) {
	128	f->open_object_section("push_progress");
	129	f->dump_stream("pushing_to") << j->first;
	130	{
	131	f->open_object_section("push_info");
	132	j->second.dump(f);
	133	f->close_section();
	134	}
	135	f->close_section();
	136	}
	137	f->close_section();
	138	}
	139	f->close_section();
	140	}
	141	f->close_section();
	142	}
	143	}
	144
	145	int objects_read_sync(
	146	const hobject_t &hoid,
	147	uint64_t off,
	148	uint64_t len,
	149	uint32_t op_flags,
	150	bufferlist *bl) override;
	151
9f95a23c TL	152	int objects_readv_sync(
	153	const hobject_t &hoid,
	154	map<uint64_t, uint64_t>&& m,
	155	uint32_t op_flags,
	156	bufferlist *bl) override;
	157
7c673cae FG	158	void objects_read_async(
	159	const hobject_t &hoid,
	160	const list<pair<boost::tuple<uint64_t, uint64_t, uint32_t>,
	161	pair<bufferlist, Context> > > &to_read,
	162	Context *on_complete,
	163	bool fast_read = false) override;
	164
	165	private:
	166	// push
	167	struct PushInfo {
	168	ObjectRecoveryProgress recovery_progress;
	169	ObjectRecoveryInfo recovery_info;
	170	ObjectContextRef obc;
	171	object_stat_sum_t stat;
	172	ObcLockManager lock_manager;
	173
	174	void dump(Formatter *f) const {
	175	{
	176	f->open_object_section("recovery_progress");
	177	recovery_progress.dump(f);
	178	f->close_section();
	179	}
	180	{
	181	f->open_object_section("recovery_info");
	182	recovery_info.dump(f);
	183	f->close_section();
	184	}
	185	}
	186	};
	187	map<hobject_t, map<pg_shard_t, PushInfo>> pushing;
	188
	189	// pull
	190	struct PullInfo {
	191	pg_shard_t from;
	192	hobject_t soid;
	193	ObjectRecoveryProgress recovery_progress;
	194	ObjectRecoveryInfo recovery_info;
	195	ObjectContextRef head_ctx;
	196	ObjectContextRef obc;
	197	object_stat_sum_t stat;
	198	bool cache_dont_need;
	199	ObcLockManager lock_manager;
	200
	201	void dump(Formatter *f) const {
	202	{
	203	f->open_object_section("recovery_progress");
	204	recovery_progress.dump(f);
	205	f->close_section();
	206	}
	207	{
	208	f->open_object_section("recovery_info");
	209	recovery_info.dump(f);
	210	f->close_section();
	211	}
	212	}
	213
	214	bool is_complete() const {
	215	return recovery_progress.is_complete(recovery_info);
	216	}
	217	};
	218
	219	map<hobject_t, PullInfo> pulling;
	220
11fdf7f2	221	// Reverse mapping from osd peer to objects being pulled from that peer
7c673cae FG	222	map<pg_shard_t, set<hobject_t> > pull_from_peer;
	223	void clear_pull(
	224	map<hobject_t, PullInfo>::iterator piter,
	225	bool clear_pull_from_peer = true);
	226	void clear_pull_from(
	227	map<hobject_t, PullInfo>::iterator piter);
	228
	229	void _do_push(OpRequestRef op);
	230	void _do_pull_response(OpRequestRef op);
	231	void do_push(OpRequestRef op) {
	232	if (is_primary()) {
	233	_do_pull_response(op);
	234	} else {
	235	_do_push(op);
	236	}
	237	}
	238	void do_pull(OpRequestRef op);
	239	void do_push_reply(OpRequestRef op);
	240
	241	bool handle_push_reply(pg_shard_t peer, const PushReplyOp &op, PushOp *reply);
	242	void handle_pull(pg_shard_t peer, PullOp &op, PushOp *reply);
	243
	244	struct pull_complete_info {
	245	hobject_t hoid;
	246	object_stat_sum_t stat;
	247	};
	248	bool handle_pull_response(
	249	pg_shard_t from, const PushOp &op, PullOp *response,
	250	list<pull_complete_info> *to_continue,
	251	ObjectStore::Transaction *t);
	252	void handle_push(pg_shard_t from, const PushOp &op, PushReplyOp *response,
11fdf7f2	253	ObjectStore::Transaction *t, bool is_repair);
7c673cae FG	254
	255	static void trim_pushed_data(const interval_set<uint64_t> &copy_subset,
	256	const interval_set<uint64_t> &intervals_received,
	257	bufferlist data_received,
	258	interval_set<uint64_t> *intervals_usable,
	259	bufferlist *data_usable);
224ce89b	260	void _failed_pull(pg_shard_t from, const hobject_t &soid);
7c673cae FG	261
	262	void send_pushes(int prio, map<pg_shard_t, vector<PushOp> > &pushes);
	263	void prep_push_op_blank(const hobject_t& soid, PushOp *op);
	264	void send_pulls(
	265	int priority,
	266	map<pg_shard_t, vector<PullOp> > &pulls);
	267
	268	int build_push_op(const ObjectRecoveryInfo &recovery_info,
	269	const ObjectRecoveryProgress &progress,
	270	ObjectRecoveryProgress *out_progress,
	271	PushOp *out_op,
	272	object_stat_sum_t *stat = 0,
	273	bool cache_dont_need = true);
	274	void submit_push_data(const ObjectRecoveryInfo &recovery_info,
	275	bool first,
	276	bool complete,
9f95a23c	277	bool clear_omap,
7c673cae	278	bool cache_dont_need,
9f95a23c	279	interval_set<uint64_t> &data_zeros,
7c673cae FG	280	const interval_set<uint64_t> &intervals_included,
	281	bufferlist data_included,
	282	bufferlist omap_header,
	283	const map<string, bufferlist> &attrs,
	284	const map<string, bufferlist> &omap_entries,
	285	ObjectStore::Transaction *t);
	286	void submit_push_complete(const ObjectRecoveryInfo &recovery_info,
	287	ObjectStore::Transaction *t);
	288
	289	void calc_clone_subsets(
	290	SnapSet& snapset, const hobject_t& poid, const pg_missing_t& missing,
	291	const hobject_t &last_backfill,
	292	interval_set<uint64_t>& data_subset,
	293	map<hobject_t, interval_set<uint64_t>>& clone_subsets,
	294	ObcLockManager &lock_manager);
	295	void prepare_pull(
	296	eversion_t v,
	297	const hobject_t& soid,
	298	ObjectContextRef headctx,
	299	RPGHandle *h);
	300	int start_pushes(
	301	const hobject_t &soid,
	302	ObjectContextRef obj,
	303	RPGHandle *h);
224ce89b	304	int prep_push_to_replica(
7c673cae FG	305	ObjectContextRef obc, const hobject_t& soid, pg_shard_t peer,
7c673cae FG	306	PushOp *pop, bool cache_dont_need = true);
224ce89b	307	int prep_push(
7c673cae FG	308	ObjectContextRef obc,
	309	const hobject_t& oid, pg_shard_t dest,
	310	PushOp *op,
	311	bool cache_dont_need);
224ce89b	312	int prep_push(
7c673cae FG	313	ObjectContextRef obc,
	314	const hobject_t& soid, pg_shard_t peer,
	315	eversion_t version,
	316	interval_set<uint64_t> &data_subset,
	317	map<hobject_t, interval_set<uint64_t>>& clone_subsets,
	318	PushOp *op,
	319	bool cache,
	320	ObcLockManager &&lock_manager);
	321	void calc_head_subsets(
	322	ObjectContextRef obc, SnapSet& snapset, const hobject_t& head,
	323	const pg_missing_t& missing,
	324	const hobject_t &last_backfill,
	325	interval_set<uint64_t>& data_subset,
	326	map<hobject_t, interval_set<uint64_t>>& clone_subsets,
	327	ObcLockManager &lock_manager);
	328	ObjectRecoveryInfo recalc_subsets(
	329	const ObjectRecoveryInfo& recovery_info,
	330	SnapSetContext *ssc,
	331	ObcLockManager &lock_manager);
	332
	333	/**
	334	* Client IO
	335	*/
11fdf7f2	336	struct InProgressOp : public RefCountedObject {
7c673cae FG	337	ceph_tid_t tid;
7c673cae FG	338	set<pg_shard_t> waiting_for_commit;
7c673cae	339	Context *on_commit;
7c673cae FG	340	OpRequestRef op;
7c673cae FG	341	eversion_t v;
7c673cae	342	bool done() const {
11fdf7f2	343	return waiting_for_commit.empty();
7c673cae	344	}
9f95a23c TL	345	private:
	346	FRIEND_MAKE_REF(InProgressOp);
	347	InProgressOp(ceph_tid_t tid, Context *on_commit, OpRequestRef op, eversion_t v)
	348	:
	349	tid(tid), on_commit(on_commit),
	350	op(op), v(v) {}
7c673cae	351	};
9f95a23c	352	map<ceph_tid_t, ceph::ref_t<InProgressOp>> in_progress_ops;
7c673cae FG	353	public:
7c673cae FG	354	friend class C_OSD_OnOpCommit;
7c673cae FG	355
	356	void call_write_ordered(std::function<void(void)> &&cb) override {
	357	// ReplicatedBackend submits writes inline in submit_transaction, so
	358	// we can just call the callback.
	359	cb();
	360	}
	361
	362	void submit_transaction(
	363	const hobject_t &hoid,
	364	const object_stat_sum_t &delta_stats,
	365	const eversion_t &at_version,
	366	PGTransactionUPtr &&t,
	367	const eversion_t &trim_to,
9f95a23c	368	const eversion_t &min_last_complete_ondisk,
7c673cae	369	const vector<pg_log_entry_t> &log_entries,
9f95a23c	370	std::optional<pg_hit_set_history_t> &hset_history,
7c673cae FG	371	Context *on_all_commit,
	372	ceph_tid_t tid,
	373	osd_reqid_t reqid,
	374	OpRequestRef op
	375	) override;
	376
	377	private:
	378	Message * generate_subop(
	379	const hobject_t &soid,
	380	const eversion_t &at_version,
	381	ceph_tid_t tid,
	382	osd_reqid_t reqid,
	383	eversion_t pg_trim_to,
9f95a23c	384	eversion_t min_last_complete_ondisk,
7c673cae FG	385	hobject_t new_temp_oid,
7c673cae FG	386	hobject_t discard_temp_oid,
11fdf7f2	387	const bufferlist &log_entries,
9f95a23c	388	std::optional<pg_hit_set_history_t> &hset_history,
7c673cae FG	389	ObjectStore::Transaction &op_t,
	390	pg_shard_t peer,
	391	const pg_info_t &pinfo);
	392	void issue_op(
	393	const hobject_t &soid,
	394	const eversion_t &at_version,
	395	ceph_tid_t tid,
	396	osd_reqid_t reqid,
	397	eversion_t pg_trim_to,
9f95a23c	398	eversion_t min_last_complete_ondisk,
7c673cae FG	399	hobject_t new_temp_oid,
	400	hobject_t discard_temp_oid,
	401	const vector<pg_log_entry_t> &log_entries,
9f95a23c	402	std::optional<pg_hit_set_history_t> &hset_history,
7c673cae FG	403	InProgressOp *op,
7c673cae FG	404	ObjectStore::Transaction &op_t);
9f95a23c	405	void op_commit(const ceph::ref_t<InProgressOp>& op);
7c673cae FG	406	void do_repop_reply(OpRequestRef op);
	407	void do_repop(OpRequestRef op);
	408
	409	struct RepModify {
	410	OpRequestRef op;
11fdf7f2	411	bool committed;
7c673cae FG	412	int ackerosd;
	413	eversion_t last_complete;
	414	epoch_t epoch_started;
	415
	416	ObjectStore::Transaction opt, localt;
	417
11fdf7f2	418	RepModify() : committed(false), ackerosd(-1),
7c673cae FG	419	epoch_started(0) {}
7c673cae FG	420	};
11fdf7f2	421	typedef std::shared_ptr<RepModify> RepModifyRef;
7c673cae	422
7c673cae FG	423	struct C_OSD_RepModifyCommit;
7c673cae FG	424
7c673cae	425	void repop_commit(RepModifyRef rm);
11fdf7f2	426	bool auto_repair_supported() const override { return store->has_builtin_csum(); }
7c673cae FG	427
7c673cae FG	428
28e407b8 AA	429	int be_deep_scrub(
	430	const hobject_t &poid,
	431	ScrubMap &map,
	432	ScrubMapBuilder &pos,
	433	ScrubMap::object &o) override;
7c673cae FG	434	uint64_t be_get_ondisk_size(uint64_t logical_size) override { return logical_size; }
	435	};
	436
	437	#endif