技术分享 | OceanBase写入限速源码解读

作者：陈慧明

爱可生测试团队成员，主要参与dmp和dble自动化测试项目。

本文来源：原创投稿

* 爱可生开源社区出品，原创内容未经授权不得随意使用，转载请联系小编并注明来源。

一、简介

OceanBase中的写入限速机制旨在控制系统中写入操作（一般写入操作包括插入、更新和删除等）的速率，目的是为了提高数据库系统的稳定性。本文主要通过以下2个参数来解释写入限速的实现机制。

1. writing_throttling_trigger_percentage：设置写入速度的阈值百分比。当内存使用达到该阈值时，触发写入限速机制。默认值为60，取值范围为1到100（100表示关闭写入限速）。
2. writing_throttling_maximum_duration：指定触发写入限速后，所需的剩余内存分配时间。默认值为2小时。通常情况下，不需要修改该参数。

请注意，OceanBase 2.2.30 及之后版本才开始支持该机制。

二、实现原理

1. 进入限速逻辑

当执行写入操作申请内存时，触发写入限速条件：已使用的 Memstore 内存超过设定的比例（比例阈值由 writing_throttling_trigger_percentage 参数确定），系统进入限速逻辑。

2. 多次限速

限速逻辑会将本次申请内存的任务分成多次进行限速。每次限速的执行时间最多为20毫秒。

3.系统在每次限速中进行一个限速循环，在限速循环中，系统会检查以下条件：

• 内存释放：系统检查内存是否已经释放足够多的内存（满足不进入限速的条件），系统已经不需要限速。
• SQL执行时间限制：系统检查SQL的执行时间是否已经达到限制。如果已经达到限制，则系统退出限速循环，并将SQL完成的信息发送给客户端。
• 休眠时间：系统检查是否已经休眠了20秒。如果已经休眠，则系统退出限速循环，并将SQL完成的信息发送给客户端。

4. 完成限速

如果上述任意一项条件满足，系统将退出限速循环，并将SQL完成的信息发送给客户端。这样可以确保SQL能够成功执行完成，并保证系统的稳定性。

5. 流程参考

三、源码解读

int ObTablet::insert_row_without_rowkey_check(
    ObRelativeTable &relative_table,
    ObStoreCtx &store_ctx,
    const common::ObIArray<share::schema::ObColDesc> &col_descs,
    const storage::ObStoreRow &row)
{
  int ret = OB_SUCCESS;
  {
    // insert_row_without_rowkey_check 执行结束时会调用ObStorageTableGuard的析构函数,进行限速处理
    ObStorageTableGuard guard(this, store_ctx, true);
    ObMemtable *write_memtable = nullptr;
    ...
    //write_memtable->set()会调用ObFifoArena::alloc()申请内存, 在分配内存时进行限速判断
    else if (OB_FAIL(write_memtable->set(store_ctx, relative_table.get_table_id(),
    full_read_info_, col_descs, row)))
    ...
  }
  return ret;
}

该方法会实例化 ObStorageTableGuard类 , 限速的执行过程定义在该类的析构函数内, 所以程序会在执行完 write_memtable 后才进行限速。后续会进行写Memtable的流程,这里不做赘述, 大致调用堆栈如下：

| > oceanbase::storage::ObTablet::insert_row_without_rowkey_check(...) (/src/storage/tablet/ob_tablet.cpp:1425)
| + > oceanbase::memtable::ObMemtable::set(...) (/src/storage/memtable/ob_memtable.cpp:339)
| + - > oceanbase::memtable::ObMemtable::set_(...) (/src/storage/memtable/ob_memtable.cpp:2538)
| + - x > oceanbase::memtable::ObMemtable::mvcc_write_(...) (/src/storage/memtable/ob_memtable.cpp:2655)
| + - x = > oceanbase::memtable::ObMvccEngine::create_kv(...) (/src/storage/memtable/mvcc/ob_mvcc_engine.cpp:275)
| + - x = | > oceanbase::memtable::ObMTKVBuilder::dup_key(...) (/src/storage/memtable/ob_memtable.h:77)
| + - x = | + > oceanbase::common::ObGMemstoreAllocator::AllocHandle::alloc(...) (/src/share/allocator/ob_gmemstore_allocator.h:84)
| + - x = | + - > oceanbase::common::ObGMemstoreAllocator::alloc(...) (/src/share/allocator/ob_gmemstore_allocator.cpp:125)
| + - x = | + - x > oceanbase::common::ObFifoArena::alloc(...) (/src/share/allocator/ob_fifo_arena.cpp:157)
| + - x = | + - x = > oceanbase::common::ObFifoArena::speed_limit(...)(/src/share/allocator/ob_fifo_arena.cpp:301)
| + - x = | + - x = | > oceanbase::common::ObFifoArena::ObWriteThrottleInfo::check_and_calc_decay_factor(...)(/src/share/allocator/ob_fifo_arena.cpp:75)
| + > oceanbase::storage::ObStorageTableGuard::~ObStorageTableGuard(...) (/src/storage/ob_storage_table_guard.cpp:53)

ObFifoArena::alloc()

写memtable时会申请内存, 这时候会去判断是否需要限速

void* ObFifoArena::alloc(int64_t adv_idx, Handle& handle, int64_t size)
{
  int ret = OB_SUCCESS;
  void* ptr = NULL;
  int64_t rsize = size + sizeof(Page) + sizeof(Ref);
  // 调用speed limit 判断限速
  speed_limit(ATOMIC_LOAD(&hold_), size);
  ...
}

ObFifoArena::speed_limit()

这个方法主要用来判断是否需要限速，同时根据配置的writing_throttling_maximum_duration值，计算出一个衰减因子用于等待时间的计算

void ObFifoArena::speed_limit(const int64_t cur_mem_hold, const int64_t alloc_size)
{
  int ret = OB_SUCCESS;
  //获取租户的writing_throttling_trigger_percentage值
  int64_t trigger_percentage = get_writing_throttling_trigger_percentage_();
  int64_t trigger_mem_limit = 0;
  bool need_speed_limit = false;
  int64_t seq = 0;
  int64_t throttling_interval = 0;
  // trigger_percentage <100 ,表示开启限速,再进行内存使用是否达到触发阈值的判断
  if (trigger_percentage < 100) {
    if (OB_UNLIKELY(cur_mem_hold < 0 || alloc_size <= 0 || lastest_memstore_threshold_ <= 0 || trigger_percentage <= 0)) {
      COMMON_LOG(ERROR, "invalid arguments", K(cur_mem_hold), K(alloc_size), K(lastest_memstore_threshold_), K(trigger_percentage));
    } else if (cur_mem_hold > (trigger_mem_limit = lastest_memstore_threshold_ * trigger_percentage / 100)) {
      // 当前使用内存超过触发阈值,需要限速,设置need_speed_limit 为true
      need_speed_limit = true;
      // 获取writing_throttling_maximum_duration的值,默认 2h
      int64_t alloc_duration = get_writing_throttling_maximum_duration_();
      // 计算衰减因子,用于sleep时间计算
      if (OB_FAIL(throttle_info_.check_and_calc_decay_factor(lastest_memstore_threshold_, trigger_percentage, alloc_duration))) {
        COMMON_LOG(WARN, "failed to check_and_calc_decay_factor", K(cur_mem_hold), K(alloc_size), K(throttle_info_));
      }
    }
  
    //这块代码是将内存和时钟值绑定,确保内存分配和写入限速的稳定性
    advance_clock();
    seq = ATOMIC_AAF(&max_seq_, alloc_size);
    get_seq() = seq;
     
    // 将need_speed_limit 赋值给tl_need_speed_limit 线程变量
    tl_need_speed_limit() = need_speed_limit;
    //日志记录,限速信息
    if (need_speed_limit && REACH_TIME_INTERVAL(1 * 1000 * 1000L)) {
      COMMON_LOG(INFO, "report write throttle info", K(alloc_size), K(attr_), K(throttling_interval),
                  "max_seq_", ATOMIC_LOAD(&max_seq_), K(clock_),
                  K(cur_mem_hold), K(throttle_info_), K(seq));
    }
  }
}

计算衰减因子

int ObFifoArena::ObWriteThrottleInfo::check_and_calc_decay_factor(int64_t memstore_threshold,
                                                                  int64_t trigger_percentage,
                                                                  int64_t alloc_duration)
{
  int ret = OB_SUCCESS;
  if (memstore_threshold != memstore_threshold_
      || trigger_percentage != trigger_percentage_
      || alloc_duration != alloc_duration_
      || decay_factor_ <= 0) {
    memstore_threshold_ = memstore_threshold;
    trigger_percentage_ = trigger_percentage;
    alloc_duration_ = alloc_duration;
    int64_t available_mem = (100 - trigger_percentage_) * memstore_threshold_ / 100;
    double N =  static_cast<double>(available_mem) / static_cast<double>(MEM_SLICE_SIZE);
    decay_factor_ = (static_cast<double>(alloc_duration) - N * static_cast<double>(MIN_INTERVAL))/ static_cast<double>((((N*(N+1)*N*(N+1)))/4));
    decay_factor_ = decay_factor_ < 0 ? 0 : decay_factor_;
    COMMON_LOG(INFO, "recalculate decay factor", K(memstore_threshold_), K(trigger_percentage_),
               K(decay_factor_), K(alloc_duration), K(available_mem), K(N));
  }
  return ret;
}

decay_factor公式中，alloc_duration为writing_throttling_maximum_duration的值，4.0版本中为2h，MIN_INTERVAL默认值20ms。

简单来说，这个衰减因子是根据当前可用内存和writing_throttling_maximum_duration的值通过一个多项式计算出来的，整个过程如果writing_throttling_maximum_duration值不做调整，每次休眠时间会随着可用内存逐渐减少而慢慢增加。

限速流程执行

ObStorageTableGuard::~ObStorageTableGuard()
{
  //tl_need_speed_limit 在ObFifoArena::alloc()方法中赋值
  bool &need_speed_limit = tl_need_speed_limit();
  // 在写操作的上下文中, 创建ObStorageTableGuard 实例时，need_control_mem_ 会被赋值为true
  if (need_control_mem_ && need_speed_limit) {
    bool need_sleep = true;
    int64_t left_interval = SPEED_LIMIT_MAX_SLEEP_TIME;
    //SPEED_LIMIT_MAX_SLEEP_TIME 默认20s,表示最大sleep时间
    if (!for_replay_) {
        // 如果不是回放日志
        //store_ctx_.timeout_ - ObTimeUtility::current_time() 表示距离事务超时还要多久,如果该值小于0,表示事务已经超时
        //两者取小
      left_interval = min(SPEED_LIMIT_MAX_SLEEP_TIME, store_ctx_.timeout_ - ObTimeUtility::current_time());
    }
    // 如果memtable是冻结状态,不需要限速
    if (NULL != memtable_) {
      need_sleep = memtable_->is_active_memtable();
    }
    uint64_t timeout = 10000;//10s
    //事件记录, 可以在v$session_event中查看,event名: memstore memory page alloc wait
    //可以通过sql: select * from v$session_event where EVENT='memstore memory page alloc wait' 查询;
    common::ObWaitEventGuard wait_guard(common::ObWaitEventIds::MEMSTORE_MEM_PAGE_ALLOC_WAIT, timeout, 0, 0, left_interval);
 
    reset();
    int tmp_ret = OB_SUCCESS;
    bool has_sleep = false;
    int64_t sleep_time = 0;
    int time = 0;
    int64_t &seq = get_seq();
    if (store_ctx_.mvcc_acc_ctx_.is_write()) {
      ObGMemstoreAllocator* memstore_allocator = NULL;
      //获取当前租户的memstore内存分配器
      if (OB_SUCCESS != (tmp_ret = ObMemstoreAllocatorMgr::get_instance().get_tenant_memstore_allocator(
          MTL_ID(), memstore_allocator))) {
      } else if (OB_ISNULL(memstore_allocator)) {
        LOG_WARN_RET(OB_ALLOCATE_MEMORY_FAILED, "get_tenant_mutil_allocator failed", K(store_ctx_.tablet_id_), K(tmp_ret));
      } else {
        while (need_sleep &&
               !memstore_allocator->check_clock_over_seq(seq) &&
               (left_interval > 0)) {
          if (for_replay_) {
            // 如果是回放日志,并且当前租户下有正在进行的日志流,不做休眠,直接break
            if(MTL(ObTenantFreezer *)->exist_ls_freezing()) {
              break;
            }
          }
          //计算休眠时间
          int64_t expected_wait_time = memstore_allocator->expected_wait_time(seq);
          if (expected_wait_time == 0) {
            break;
          }
          //SLEEP_INTERVAL_PER_TIME 单次休眠时间,默认20ms
          //线程休眠,每次最多20ms
          uint32_t sleep_interval =
            static_cast<uint32_t>(min(min(left_interval, SLEEP_INTERVAL_PER_TIME), expected_wait_time));
          ::usleep(sleep_interval);
          // 累加休眠时间
          sleep_time += sleep_interval;
          // 休眠次数
          time++;
          //每次休眠之后,减去休眠时间
          left_interval -= sleep_interval;
          has_sleep = true;
          //每次休眠之后,重新判断是否需要限速,因为可能在休眠过程中,内存经过转储后已经释放出来了,这时候就不需要继续限速了
          need_sleep = memstore_allocator->need_do_writing_throttle();
        }
      }
    }
    // 日志记录,限速执行详情
    if (REACH_TIME_INTERVAL(100 * 1000L) &&
        sleep_time > 0) {
      int64_t cost_time = ObTimeUtility::current_time() - init_ts_;
      LOG_INFO("throttle situation", K(sleep_time), K(time), K(seq), K(for_replay_), K(cost_time));
    }
 
    if (for_replay_ && has_sleep) {
      get_replay_is_writing_throttling() = true;
    }
  }
  reset();
}

OB的写入限速功能是在ObStorageTableGuard类的析构函数中实现的。由于该函数会在memtable写入完成后才被调用，因此限速行为是后置的，会影响下一次内存分配。换言之，在当前写入操作完成后，才会判断是否需要执行限速，若需要，会延迟下一次内存分配。这种设计既确保限速不会影响当前的写入操作，又能有效控制内存的分配和消耗。

四、使用方法

该参数是租户级别的参数，可以在租户管理员账号下或者在sys租户中指定租户，设置内存写入达到 80% 开始限速，并保证剩余内存足够提供 2h 的写入限速，示例:

obclient> ALTER SYSTEM SET writing_throttling_trigger_percentage = 80;
Query OK, 0 rows affected
obclient> ALTER SYSTEM SET writing_throttling_maximum_duration = '2h';
Query OK, 0 rows affected
  
或者在sys租户中指定租户
obclient> ALTER SYSTEM SET writing_throttling_trigger_percentage = 80 tenant=<tenant_name>;

五、使用场景

select * from v$session_event where EVENT='memstore memory page alloc wait' \G;
*************************** 94. row ***************************
           CON_ID: 1
           SVR_IP: 10.186.64.124
         SVR_PORT: 22882
              SID: 3221487713
            EVENT: memstore memory page alloc wait
      TOTAL_WAITS: 182673
   TOTAL_TIMEOUTS: 0
      TIME_WAITED: 1004.4099
     AVERAGE_WAIT: 0.005498403704981032
         MAX_WAIT: 12.3022
TIME_WAITED_MICRO: 10044099
              CPU: NULL
         EVENT_ID: 11015
    WAIT_CLASS_ID: 109
      WAIT_CLASS#: 9
       WAIT_CLASS: SYSTEM_IO

[2023-04-17 17:17:30.695621] INFO  [COMMON] speed_limit (ob_fifo_arena.cpp:319) [26466][T1_L0_G0][T1][Y59620ABA407C-0005F9818D1BFE06-0-0] [lt=2] report write throttle info(alloc_size=32, attr_=tenant_id=1, label=Memstore, ctx_id=1, prio=0, throttling_interval=0, max_seq_=11045142952, clock_=11045143112, cur_mem_hold=524288000, throttle_info_={decay_factor_:"6.693207379708156213e-02", alloc_duration_:7200000000, trigger_percentage_:21, memstore_threshold_:2147483600, period_throttled_count_:0, period_throttled_time_:0, total_throttled_count_:0, total_throttled_time_:0}, seq=11045142952)

同时grep 'throttle situation' observer.log，可以看到这次限速的具体内容。

[2023-04-17 17:17:31.006880] INFO  [STORAGE] ~ObStorageTableGuard (ob_storage_table_guard.cpp:109) [26466][T1_L0_G0][T1][Y59620ABA407C-0005F9818D1BFE06-0-0] [lt=85] throttle situation(sleep_time=4, time=1, seq=11048795064, for_replay_=false, cost_time=7025)