Ceph►OSD

PG 存储调用与数据读取机制分析

PG 存储调用与数据读取机制分析

1. 概述

PG（Placement Group）是 Ceph OSD 中管理对象存储的核心组件。PG 通过 ObjectStore 接口与底层存储后端（如 BlueStore、FileStore）交互，实现对象的读写、元数据管理等功能。

1.1 核心组件

  • ObjectStore：存储后端抽象接口
  • CollectionHandle (ch)：集合句柄，用于标识和管理 PG 的存储集合
  • PGTransaction：PG 层的事务抽象
  • ObjectStore::Transaction：存储层的事务实现

1.2 存储层次

1
2
3
4
5
6
7
PG (PrimaryLogPG)
    ↓
PGBackend (ReplicatedBackend/ECBackend)
    ↓
ObjectStore (BlueStore/FileStore)
    ↓
底层存储设备

2. 存储初始化与集合管理

2.1 PG 存储初始化

PG 在创建时会初始化存储相关的组件：

1
2
3
4
5
6
7
8
9
10
11
12
13
// PG 构造函数
PG::PG(OSDService *o, OSDMapRef curmap, const PGPool &_pool, spg_t p) :
  pg_whoami(o->whoami, p.shard),
  pg_id(p),
  coll(p),  // 集合 ID，基于 PG ID
  osd(o),
  cct(o->cct),
  osdriver(osd->store, coll_t(), OSD::make_snapmapper_oid()),
  // ...
{
  // coll 是 PG 的集合标识符
  // 格式：coll_t(pgid)，例如 coll_t(1.23)
}

关键成员变量：

  • coll：集合 ID（coll_t），标识 PG 的存储集合
  • ch：集合句柄（ObjectStore::CollectionHandle），用于访问集合

2.2 打开集合（Collection）

PG 在加载时会打开对应的存储集合：

1
2
// OSD::load_pgs() 中
pg->ch = store->open_collection(pg->coll);

集合句柄的作用：

  • 标识 PG 的存储集合
  • 用于事务提交和查询操作
  • 保证事务在同一个集合内按顺序执行

2.3 集合与 PG 的对应关系

1
2
3
4
5
6
// PG ID 到集合 ID 的转换
coll_t coll(pgid);  // 直接使用 PG ID 作为集合 ID

// 例如：
// pgid = spg_t(1.23)
// coll = coll_t(1.23)

3. 存储读取机制

3.1 读取 PG 元数据

PG 启动时需要从存储中读取元数据：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
// PG::read_state() - 读取 PG 状态
void PG::read_state(ObjectStore *store)
{
  // 1. 读取 PG 信息
  int r = read_info(store, pg_id, coll, info_from_disk, 
                    past_intervals_from_disk, info_struct_v);
  
  // 2. 读取 PGLog
  recovery_state.get_pg_log().read_log_and_missing(
    store, coll, pgmeta_oid, info_from_disk, 
    past_intervals_from_disk, cct);
  
  // 3. 初始化 PeeringState
  recovery_state.init(...);
}

读取 PG 信息：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
int PG::read_info(ObjectStore *store, spg_t pgid, const coll_t &coll,
                  pg_info_t &info, PastIntervals &past_intervals,
                  __u8 &struct_v)
{
  // 1. 打开集合
  auto ch = store->open_collection(coll);
  ceph_assert(ch);
  
  // 2. 准备要读取的键
  set<string> keys;
  keys.insert(string(infover_key));  // 版本信息
  keys.insert(string(info_key));      // PG 信息
  keys.insert(string(biginfo_key));  // 大信息（PastIntervals）
  keys.insert(string(fastinfo_key));  // 快速信息
  
  // 3. 构建元数据对象 ID
  ghobject_t pgmeta_oid(pgid.make_pgmeta_oid());
  
  // 4. 从 OMAP 读取值
  map<string,bufferlist> values;
  int r = store->omap_get_values(ch, pgmeta_oid, keys, &values);
  
  // 5. 解码数据
  auto p = values[string(infover_key)].cbegin();
  decode(struct_v, p);
  
  p = values[string(info_key)].begin();
  decode(info, p);
  
  p = values[string(biginfo_key)].begin();
  decode(past_intervals, p);
  
  return 0;
}

3.2 读取对象数据

3.2.1 通过 PGBackend 读取

PG 通过 PGBackend 读取对象数据：

1
2
3
4
5
6
7
8
9
10
11
12
// PrimaryLogPG::do_op() - 处理客户端操作
void PrimaryLogPG::do_op(OpRequestRef& op)
{
  // 1. 获取对象上下文
  ObjectContextRef obc = get_object_context(head, false);
  
  // 2. 创建操作上下文
  OpContext *ctx = new OpContext(op, obc, this);
  
  // 3. 执行操作（包括读取）
  execute_ctx(ctx);
}

3.2.2 读取操作流程

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
// PrimaryLogPG::execute_ctx() - 执行操作上下文
void PrimaryLogPG::execute_ctx(OpContext *ctx)
{
  // 1. 准备事务
  int result = prepare_transaction(ctx);
  
  // 2. 如果是只读操作
  if (ctx->op_t->empty() || result < 0) {
    // 完成读取
    complete_read_ctx(result, ctx);
    return;
  }
  
  // 3. 写入操作需要提交事务
  // ...
}

3.2.3 直接存储读取

在某些场景下，PG 会直接调用存储接口读取数据：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
// 读取对象属性
int r = osd->store->getattr(
    ch,
    ghobject_t(oid, ghobject_t::NO_GEN, pg_whoami.shard),
    OI_ATTR,  // 对象信息属性
    bv);

// 读取对象统计信息
struct stat st;
int r = osd->store->stat(
    ch,
    ghobject_t(oid, ghobject_t::NO_GEN, pg_whoami.shard),
    &st);

// 读取 OMAP 值
int r = store->omap_get_values(
    ch,
    pgmeta_oid,
    keys,
    &values);

3.3 异步读取

对于 EC（纠删码）池，支持异步读取：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
// PrimaryLogPG::OpContext::start_async_reads()
void PrimaryLogPG::OpContext::start_async_reads(PrimaryLogPG *pg)
{
  inflightreads = 1;
  
  // 准备异步读取请求
  list<boost::tuple<hobject_t, uint64_t, uint64_t, uint32_t>> in;
  in.swap(pending_async_reads);
  
  // 通过 PGBackend 异步读取
  pg->pgbackend->objects_read_async(
    in,
    new OnReadComplete(pg, this),
    pg->get_pool().fast_read);
}

4. 存储写入机制

4.1 事务构建

PG 通过事务（Transaction）来组织写入操作：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
// PrimaryLogPG::execute_ctx() - 构建事务
void PrimaryLogPG::execute_ctx(OpContext *ctx)
{
  // 1. 创建 PG 事务
  ctx->op_t.reset(new PGTransaction);
  
  // 2. 准备事务（添加操作）
  int result = prepare_transaction(ctx);
  
  // 3. 将 PG 事务转换为存储事务
  ObjectStore::Transaction t;
  ctx->op_t->apply(&osdriver, &t);
  
  // 4. 提交事务
  int r = osd->store->queue_transaction(ch, std::move(t), NULL);
}

4.2 事务提交流程

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
// PrimaryLogPG::execute_ctx() - 提交事务
void PrimaryLogPG::execute_ctx(OpContext *ctx)
{
  // 1. 准备存储事务
  ObjectStore::Transaction t;
  
  // 2. 应用 PG 事务到存储事务
  ctx->op_t->apply(&osdriver, &t);
  
  // 3. 注册提交回调
  ctx->register_on_commit([...]() {
    // 发送回复给客户端
    osd->send_message_osd_client(reply, m->get_connection());
  });
  
  // 4. 通过 PGBackend 提交（会复制到副本）
  issue_repop(repop, ctx);
}

4.3 通过 PGBackend 写入

PG 通过 PGBackend 处理写入操作：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
// ReplicatedBackend::submit_transaction()
void ReplicatedBackend::submit_transaction(
  const hobject_t &hoid,
  const eversion_t &version,
  PGTransactionUPtr &&t,
  const eversion_t &trim_to,
  const eversion_t &roll_forward_to,
  const vector<pg_log_entry_t> &log_entries,
  boost::optional<pg_hit_set_history_t> &hset_history,
  Context *on_all_commit,
  Context *on_all_applied,
  Context *on_local_applied_sync)
{
  // 1. 转换 PG 事务为存储事务
  ObjectStore::Transaction _t;
  t->apply(&osdriver, &_t);
  
  // 2. 提交本地事务
  int r = store->queue_transaction(ch, std::move(_t), ...);
  
  // 3. 发送到副本（如果是主副本）
  if (is_primary()) {
    send_transaction_to_replicas(...);
  }
}

4.4 直接存储写入

在某些场景下，PG 会直接写入存储：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
// 写入 PG 状态
void PeeringState::write_if_dirty(ObjectStore::Transaction& t)
{
  if (dirty_info) {
    // 准备写入数据
    map<string,bufferlist> km;
    prepare_info_keymap(...);
    
    // 添加到事务
    t.omap_setkeys(coll, pgmeta_oid, km);
  }
  
  // 提交事务
  int r = store->queue_transaction(ch, std::move(t), NULL);
}

5. 存储接口详解

5.1 ObjectStore 核心接口

5.1.1 集合操作

1
2
3
4
5
6
7
8
9
10
11
// 打开集合
ObjectStore::CollectionHandle ch = store->open_collection(coll);

// 创建新集合
ObjectStore::CollectionHandle ch = store->create_new_collection(coll);

// 刷新集合（等待所有事务完成）
ch->flush();

// 异步刷新提交
bool idle = ch->flush_commit(callback);

5.1.2 对象读取

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
// 读取对象数据
int store->read(
    CollectionHandle &ch,
    const ghobject_t &oid,
    uint64_t offset,
    size_t len,
    bufferlist &bl);

// 读取对象属性
int store->getattr(
    CollectionHandle &ch,
    const ghobject_t &oid,
    const char *name,
    bufferlist &value);

// 获取对象统计信息
int store->stat(
    CollectionHandle &ch,
    const ghobject_t &oid,
    struct stat *st);

5.1.3 OMAP 操作

1
2
3
4
5
6
7
8
9
10
11
12
// 读取 OMAP 值
int store->omap_get_values(
    CollectionHandle &ch,
    const ghobject_t &oid,
    const set<string> &keys,
    map<string,bufferlist> *out);

// 读取 OMAP 头部
int store->omap_get_header(
    CollectionHandle &ch,
    const ghobject_t &oid,
    bufferlist *header);

5.1.4 事务操作

1
2
3
4
5
6
// 提交事务
int store->queue_transaction(
    CollectionHandle &ch,
    Transaction &&t,
    TrackedOpRef op = TrackedOpRef(),
    ThreadPool::TPHandle *handle = NULL);

5.2 Transaction 操作类型

存储事务支持多种操作：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
// 写入数据
t.write(oid, offset, len, data);

// 设置属性
t.setattr(oid, name, value);

// OMAP 操作
t.omap_setkeys(oid, keys);
t.omap_setheader(oid, header);
t.omap_rmkeys(oid, keys);

// 对象操作
t.touch(oid);           // 创建对象
t.remove(oid);          // 删除对象
t.clone(oid, noid);     // 克隆对象
t.clone_range(oid, noid, offset, len, dest_offset);

6. 数据读取场景

6.1 客户端读取操作

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
// PrimaryLogPG::do_op() - 处理读取操作
void PrimaryLogPG::do_op(OpRequestRef& op)
{
  // 1. 获取对象上下文
  ObjectContextRef obc = get_object_context(head, false);
  
  // 2. 创建操作上下文
  OpContext *ctx = new OpContext(op, obc, this);
  
  // 3. 执行操作
  execute_ctx(ctx);
  
  // 4. 在 prepare_transaction() 中：
  //    - 如果是读取操作，通过 do_osd_ops() 读取数据
  //    - 数据从 ObjectContext 或直接读取存储获取
}

6.2 恢复读取

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
// ReplicatedBackend::recover_object() - 恢复对象
int ReplicatedBackend::recover_object(
  const hobject_t &hoid,
  eversion_t v,
  ObjectContextRef head,
  ObjectContextRef obc,
  RecoveryHandle *_h)
{
  if (get_parent()->get_local_missing().is_missing(hoid)) {
    // 本地缺失，从副本拉取
    prepare_pull(v, hoid, head, h);
  } else {
    // 本地有，推送到副本
    start_pushes(hoid, obc, h);
  }
}

6.3 Scrub 读取

1
2
3
4
5
6
7
8
9
// 在 Scrub 过程中读取对象
// 1. 读取对象数据
store->read(ch, oid, 0, size, data);

// 2. 读取对象信息
store->getattr(ch, oid, OI_ATTR, oi_bl);

// 3. 验证数据完整性
verify_object_data(oid, data, oi);

7. 数据写入场景

7.1 客户端写入操作

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
// PrimaryLogPG::execute_ctx() - 执行写入
void PrimaryLogPG::execute_ctx(OpContext *ctx)
{
  // 1. 准备事务
  ctx->op_t.reset(new PGTransaction);
  
  // 2. 执行 OSD 操作（构建事务）
  int result = prepare_transaction(ctx);
  
  // 3. 转换为存储事务
  ObjectStore::Transaction t;
  ctx->op_t->apply(&osdriver, &t);
  
  // 4. 通过 PGBackend 提交（包含复制）
  issue_repop(repop, ctx);
}

7.2 PG 状态写入

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
// PeeringState::write_if_dirty() - 写入 PG 状态
void PeeringState::write_if_dirty(ObjectStore::Transaction& t)
{
  if (dirty_info) {
    // 准备 PG 信息
    map<string,bufferlist> km;
    prepare_info_keymap(...);
    
    // 添加到事务
    t.omap_setkeys(coll, pgmeta_oid, km);
  }
  
  if (dirty_big_info) {
    // 准备大信息（PastIntervals）
    // ...
  }
  
  // PGLog 写入
  pg_log.write_log_and_missing(t, pgmeta_oid, ...);
}

7.3 恢复写入

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
// ReplicatedBackend::run_recovery_op() - 执行恢复操作
void ReplicatedBackend::run_recovery_op(
  PGBackend::RecoveryHandle *_h,
  int priority)
{
  RPGHandle *h = static_cast<RPGHandle *>(_h);
  
  // 1. 发送推送（推送到副本）
  send_pushes(priority, h->pushes);
  
  // 2. 发送拉取（从副本拉取）
  send_pulls(priority, h->pulls);
  
  // 3. 发送删除
  send_recovery_deletes(priority, h->deletes);
}

8. 存储调用路径

8.1 读取路径

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
客户端请求
    ↓
OSD::dispatch_op()
    ↓
PrimaryLogPG::do_op()
    ↓
PrimaryLogPG::execute_ctx()
    ↓
PrimaryLogPG::prepare_transaction()
    ↓
PrimaryLogPG::do_osd_ops()  // 读取操作
    ↓
ObjectContext::get()  // 从缓存获取
    ↓ (缓存未命中)
ObjectStore::read()  // 直接读取存储
    ↓
底层存储后端（BlueStore/FileStore）

8.2 写入路径

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
客户端请求
    ↓
OSD::dispatch_op()
    ↓
PrimaryLogPG::do_op()
    ↓
PrimaryLogPG::execute_ctx()
    ↓
PrimaryLogPG::prepare_transaction()
    ↓
PGTransaction::add_*()  // 添加操作到 PG 事务
    ↓
PGTransaction::apply()  // 转换为存储事务
    ↓
PGBackend::submit_transaction()
    ↓
ObjectStore::queue_transaction()  // 提交到存储
    ↓
底层存储后端（BlueStore/FileStore）

9. 关键数据结构

9.1 CollectionHandle

1
2
3
4
5
6
7
// 集合句柄
ObjectStore::CollectionHandle ch;

// 用途：
// 1. 标识 PG 的存储集合
// 2. 用于事务提交
// 3. 保证事务顺序

9.2 PGTransaction

1
2
3
4
5
6
7
8
9
10
// PG 层事务抽象
class PGTransaction {
  // 添加操作
  void add_write(const hobject_t &oid, uint64_t off, uint64_t len, ...);
  void add_setattr(const hobject_t &oid, const string &name, ...);
  void add_omap_setkeys(const hobject_t &oid, ...);
  
  // 转换为存储事务
  void apply(OSDriver *osdriver, ObjectStore::Transaction *t);
};

9.3 ObjectStore::Transaction

1
2
3
4
5
6
7
8
9
10
// 存储层事务
class Transaction {
  // 操作列表
  vector<Op> ops;
  
  // 添加操作
  void write(const ghobject_t &oid, uint64_t off, uint64_t len, ...);
  void setattr(const ghobject_t &oid, const char *name, ...);
  void omap_setkeys(const ghobject_t &oid, ...);
};

10. 性能优化

10.1 对象上下文缓存

1
2
3
4
5
6
7
// ObjectContext 缓存对象信息
class ObjectContext {
  ObjectState obs;        // 对象状态
  SnapSetContext *ssc;    // 快照集上下文
  
  // 缓存对象信息，避免重复读取
};

10.2 批量操作

1
2
3
4
5
// 批量提交事务
vector<Transaction> tls;
tls.push_back(std::move(t1));
tls.push_back(std::move(t2));
store->queue_transactions(ch, tls, op, handle);

10.3 异步操作

1
2
3
4
5
// 异步读取（EC 池）
pgbackend->objects_read_async(
  read_requests,
  callback,
  fast_read);

11. 错误处理

11.1 读取错误

1
2
3
4
5
6
7
8
9
// 读取失败处理
int r = store->read(ch, oid, offset, len, bl);
if (r < 0) {
  if (r == -ENOENT) {
    // 对象不存在
  } else {
    // 其他错误
  }
}

11.2 写入错误

1
2
3
4
5
6
// 事务提交失败处理
int r = store->queue_transaction(ch, std::move(t), op);
if (r != 0) {
  // 处理错误
  derr << "queue_transaction failed: " << cpp_strerror(r) << dendl;
}

12. 总结

12.1 核心机制

  1. 集合管理：每个 PG 对应一个存储集合，通过 CollectionHandle 访问
  2. 事务机制：通过 PGTransaction 和 ObjectStore::Transaction 组织操作
  3. 读写分离：读取可以直接调用存储接口，写入通过事务提交
  4. 后端抽象：通过 PGBackend 抽象复制/EC 差异

12.2 关键接口

  • 集合操作：open_collection(), flush(), flush_commit()
  • 读取操作：read(), getattr(), stat(), omap_get_values()
  • 写入操作：通过 Transaction 的 write(), setattr(), omap_setkeys() 等
  • 事务提交：queue_transaction()

12.3 调用流程

  1. 初始化：PG 创建时初始化 coll，加载时打开 ch
  2. 读取：通过 ch 和 store 接口直接读取
  3. 写入：构建 PGTransaction，转换为 ObjectStore::Transaction，通过 PGBackend 提交
  4. 提交：store->queue_transaction(ch, t) 提交到底层存储

12.4 相关文件

  • PG.h/cc：PG 基础类，管理集合和存储访问
  • PrimaryLogPG.h/cc：主日志 PG 实现，处理客户端操作
  • PGBackend.h：后端抽象接口
  • ReplicatedBackend.h/cc：复制后端实现
  • ECBackend.h/cc：EC 后端实现
  • ObjectStore.h：存储后端接口定义
  • os/Transaction.h：事务定义

文章互动

阅读 --

留言

0 条留言

正在加载留言…