2025-04-30

heaptrack是kde的内存调试工具，我们在定位内存泄漏的时候，为了方便定位可以安装此工具来排查，下面是heaptrack的使用方法

安装

安装方式如下


apt install heaptrack heaptrack-gui libheaptrack

监听内存泄漏

heaptrack不需要太多的参数支持，直接heaptrack追加需要监控的进程即可，如下


heaptrack ./ukui-tablet-desktop

因为我们调试内存泄漏的时候，大概是知道泄漏的点或者行为，此时进入程序触发泄漏行为即可。当泄漏出现的时候，我们关掉程序，可以在当前目录获得名字为heaptrack.name.pid.gz类似的文件，此时直接加载此调试文件即可，如下

打开分析界面

直接使用heaptrack_gui即可，如下


heaptrack_gui /home/kylin/heaptrack.ukui-tablet-desktop.29058.gz

此时我们可以看到图片如下

这里已经大概告诉你泄漏的函数原因了，可以继续追代码了。

如果我们需要看详细的统计数据，我们可以跳转到Flame Graph中，当然任何内存泄漏都有一定阈值，我们可以设置Cost Threshold即可，这里设置10%。如下所示

这里就直接看到超过10%的Cost的调用，直接就能看出原因。

虽然上面可以查看到简单的内存泄漏，但是内存泄漏是长期的，我们需要长时间盯着内存的消耗增长，那么我们可以挂机72小时后，在Consumed中查看其增长情况，如下。

此时如果是长时间挂机，内存不是趋于平稳，那么就说明内存泄漏比较明显了。否则我们只能说内存会被一定情况下被使用。

总结

至此，我们基于heaptrack能够很方便的定位内存泄漏问题了，它不同于valgrind，大型的程序使用valgrind会导致性能十分低下，不方便排查问题。而像kde这种大型qt项目或其他，heaptrack是一个非常不错的选择。

2025-04-29

之前把rtems上支持的所有的调度器介绍了一遍，但是一直都没有说上下文是如何切换的。其实上下文切换的逻辑就是保存线程TCB和通用寄存器，然后根据新的线程TCB信息进行运行。本文从上下文切换的角度来讲清楚RTEMS中是怎么完成上下文切换的

从schedule函数角度看

对于任何的任务，调用schedule就是根据当前任务的情况执行调度，那么主动调用如下函数


( *scheduler->Operations.schedule )( scheduler, the_thread );

这里不同的调度算法实现的回调不一样，为了方便介绍，这里以优先级调度为例，那么最后调用函数如下


static inline void _Scheduler_uniprocessor_Update_heir(
  Thread_Control *heir,
  Thread_Control *new_heir
)
{
  _Assert( heir != new_heir );
#if defined(RTEMS_SMP)
  /*
   * We need this state only for _Thread_Get_CPU_time_used_locked().  Cannot
   * use _Scheduler_Thread_change_state() since THREAD_SCHEDULER_BLOCKED to
   * THREAD_SCHEDULER_BLOCKED state changes are illegal for the real SMP
   * schedulers.
   */
  heir->Scheduler.state = THREAD_SCHEDULER_BLOCKED;
  new_heir->Scheduler.state = THREAD_SCHEDULER_SCHEDULED;
#endif
  _Thread_Update_CPU_time_used( heir, _Thread_Get_CPU( heir ) );
  _Thread_Heir = new_heir;
  _Thread_Dispatch_necessary = true;
}

再设置完一些state之后，我们重点关注如下两个值的设置


  _Thread_Heir = new_heir;
  _Thread_Dispatch_necessary = true;

这里的new_heir是即将要运行的任务，他是Thread_Control * 指针。而_Thread_Dispatch_necessary是Per_CPU_Control结构体的一个成员dispatch_necessary，我们通过dispatch_necessary来判断当前任务是否要scheduler。

从_Thread_Do_dispatch函数角度看

我们当任务在合适的实际，我们可以设置dispatch enable，或者直接调用_Thread_Do_dispatch函数，这样线程就直接开始开始调度的实际行为。我们看起核心代码如下


  do {
    Thread_Control                     *heir;
    heir = _Thread_Get_heir_and_make_it_executing( cpu_self );

    if ( heir == executing ) {
      break;
    }

    _ISR_Local_enable( level );

    _Thread_Save_fp( executing );
    _Context_Switch( &executing->Registers, &heir->Registers );
    _Thread_Restore_fp( executing );
    _User_extensions_Thread_switch( NULL, executing );

    cpu_self = _Per_CPU_Get();

    _ISR_Local_disable( level );
  } while ( cpu_self->dispatch_necessary );

可以看到，只要dispatch_necessary设置为true，那么任务就会执行调度。执行调度的函数为_Context_Switch，它会根据每个架构的实现来运行上下文切换，我们基于armv8架构来介绍

从_CPU_Context_switch函数来看

这里主要是Arm64的实现，先看源码


DEFINE_FUNCTION_AARCH64(_CPU_Context_switch)
	.globl	_CPU_Context_switch_no_return
	.set	_CPU_Context_switch_no_return, _CPU_Context_switch
#ifdef AARCH64_MULTILIB_ARCH_V8_ILP32
/* Sanitize inputs for ILP32 ABI */
	mov w0, w0
	mov w1, w1
  #ifdef RTEMS_SMP
    #define reg_2 x2
  #else
    #define reg_2 w2
  #endif
#else
#define reg_2 x2
#endif

/* Start saving context */
	GET_SELF_CPU_CONTROL	reg_2
	ldr	w3, [x2, #PER_CPU_ISR_DISPATCH_DISABLE]  // 读取isr_dispatch_disable

	stp x19, x20, [x0]               // 保存aapcs中规定的callee寄存器 也就是x19-x28
	stp x21, x22, [x0, #0x10]
	stp x23, x24, [x0, #0x20]
	stp x25, x26, [x0, #0x30]
	stp x27, x28, [x0, #0x40]
	stp fp,  lr,  [x0, #0x50]       // 保存fp和lr
	mov x4,  sp                     // 保存sp寄存器
	str x4,  [x0, #0x60]

#ifdef AARCH64_MULTILIB_VFP
	add	x5, x0, #AARCH64_CONTEXT_CONTROL_D8_OFFSET
	stp d8,  d9,  [x5]              // 保存浮点寄存器
	stp d10, d11, [x5, #0x10]
	stp d12, d13, [x5, #0x20]
	stp d14, d15, [x5, #0x30]
#endif

	str	x3, [x0, #AARCH64_CONTEXT_CONTROL_ISR_DISPATCH_DISABLE] //将本线程的isr_dispatch_disable设置到x0

#ifdef RTEMS_SMP
	/*
	 * The executing thread no longer executes on this processor.  Switch
	 * the stack to the temporary interrupt stack of this processor.  Mark
	 * the context of the executing thread as not executing.
	 */
	dmb	SY          //所有数据dmb
	add	sp, x2, #(PER_CPU_INTERRUPT_FRAME_AREA + CPU_INTERRUPT_FRAME_SIZE) //分配一个栈
	mov	x3, #0  //将0写到is_executing中
	strb	w3, [x0, #AARCH64_CONTEXT_CONTROL_IS_EXECUTING_OFFSET]

.L_check_is_executing:

	/* Check the is executing indicator of the heir context */
	add	x3, x1, #AARCH64_CONTEXT_CONTROL_IS_EXECUTING_OFFSET //获取is_executing的值
	ldaxrb	w4, [x3] //原子读取x3的内容到w4
	cmp	x4, #0 
	bne	.L_get_potential_new_heir  //如果已经其他cpu执行则跳到其他线程tcb上重新判断

	/* Try to update the is executing indicator of the heir context */
	mov	x4, #1
	stlxrb	w5, w4, [x3]   // 将is_executing设置回去，结果保存在w5
	cmp	x5, #0  
	bne	.L_get_potential_new_heir // 如果设置失败，则跳到其他tcb上重新判断
	dmb	SY  // dmb所有数据
#endif

/* Start restoring context */
.L_restore:
#if !defined(RTEMS_SMP) && defined(AARCH64_MULTILIB_HAS_LOAD_STORE_EXCLUSIVE)
	clrex  // 清空独占监视器
#endif

	ldr	x3, [x1, #AARCH64_CONTEXT_CONTROL_THREAD_ID_OFFSET]  // 加载下一个线程的thread_id

	ldr	x4, [x1, #AARCH64_CONTEXT_CONTROL_ISR_DISPATCH_DISABLE] // 加载下一个线程的isr_dispatch_disable

#ifdef AARCH64_MULTILIB_VFP
	add	x5, x1, #AARCH64_CONTEXT_CONTROL_D8_OFFSET // 加载下一个线程的浮点寄存器
	ldp d8,  d9,  [x5]
	ldp d10, d11, [x5, #0x10]
	ldp d12, d13, [x5, #0x20]
	ldp d14, d15, [x5, #0x30]
#endif

	msr	TPIDR_EL0, x3   // 将x3设置到TPIDR_EL0

	str	w4, [x2, #PER_CPU_ISR_DISPATCH_DISABLE]  // 更新isr_dispatch_disable

	ldp x19, x20, [x1]  // 恢复下一个线程的callee寄存器
	ldp x21, x22, [x1, #0x10]
	ldp x23, x24, [x1, #0x20]
	ldp x25, x26, [x1, #0x30]
	ldp x27, x28, [x1, #0x40]
	ldp fp,  lr,  [x1, #0x50]   // 恢复fp和lr
	ldr x4,  [x1, #0x60]     
	mov sp,  x4       //恢复sp
	ret

上面代码尽可能给出了注释，有些补充解释如下

x0和x1

我们知道_CPU_Context_switch的原型是：


_CPU_Context_switch( _executing, _heir )

所以x0 是当前线程tcb，x1是下一个线程的tcb

Context_Control结构体

我们知道需要保存x19-x30 sp等寄存器，那么保存的地方在TPIDR_EL0上，TPIDR_EL0上存在的是Context_Control的指针，所以Context_Control结构体如下。


typedef struct {
  uint64_t register_x19;
  uint64_t register_x20;
  uint64_t register_x21;
  uint64_t register_x22;
  uint64_t register_x23;
  uint64_t register_x24;
  uint64_t register_x25;
  uint64_t register_x26;
  uint64_t register_x27;
  uint64_t register_x28;
  uint64_t register_fp;
  uint64_t register_lr;
  uint64_t register_sp;
  uint64_t isr_dispatch_disable;
  uint64_t thread_id;
#ifdef AARCH64_MULTILIB_VFP
  uint64_t register_d8;
  uint64_t register_d9;
  uint64_t register_d10;
  uint64_t register_d11;
  uint64_t register_d12;
  uint64_t register_d13;
  uint64_t register_d14;
  uint64_t register_d15;
#endif
#ifdef RTEMS_SMP
  volatile bool is_executing;
#endif
} Context_Control;

dmb SY

dmb是数据内存同步，SY是对全系统进行刷新，SY的解释如下

CLREX

Clear Exclusive access monitor，清空独占监视器，请问接下来要做独占访问，这里直接清空独占监视器

ldaxrb和stlxrb

这两个都是独占访问内存和独占存储内存，

其中stlxrb会将"<Ws>"的值返回出来，如果是0则独占访问成功，如果非0则失败

总结

至此，我们根据上下文切换的代码分析，清楚的知道了调度器在任务运行时的寄存器保存过程和sp分配过程等。相关演示操作在之前的文章有过体现，gdb能够很清楚的看到任务分配后的寄存器显示，这里就没有必要重复了。

2025-04-24

关于cbs调度算法，我们上面知道了其默认基于edf，以截止时间作为优先级，每个任务都有预算，当预算完成时任务也完成了，那么更新预算，重新周期，如果任务的unblock后的剩余占比大于设置预算设置占比，则将任务推迟运行。
本文基于rtems的测试用例，简单演示一下cbs调度算法

测试代码

为了支持测试，我们需要创建四个任务，这个比较常规，使用 rtems_task_create和rtems_task_start即可
在cbs中，我们需要额外调用rtems_cbs_initialize即可。那么代码如下


  for ( index = 1 ; index <= NUM_TASKS ; index++ ) {
    status = rtems_task_create(
      Task_name[ index ],
      Priorities[ index ],
      RTEMS_MINIMUM_STACK_SIZE * 4,
      RTEMS_DEFAULT_MODES,
      RTEMS_DEFAULT_ATTRIBUTES,
      &Task_id[ index ]
    );
    directive_failed( status, "rtems_task_create loop" );
  }

  rtems_cbs_initialize();

  for ( index = 1 ; index <= NUM_PERIODIC_TASKS ; index++ ) {
    status = rtems_task_start( Task_id[ index ], Tasks_Periodic, index );
    directive_failed( status, "rtems_task_start loop" );
  }

根据上面，任务开始时会默认调用Tasks_Periodic，我们在这里设置任务

首先，我们需要按照edf的规范，设置周期，主要调用如下


rtems_rate_monotonic_create
rtems_rate_monotonic_ident
rtems_rate_monotonic_period

对于cbs而言，我们需要预设deadline和budget，并且如果任务超限，也就是cpu利用率大于1时，我们可以设置一个函数回调，如下


params.deadline = Periods[ argument ];
params.budget = Execution[ argument ]+1;
rtems_cbs_create_server( &params, &overrun_handler_task_4, &server_id )

如果我们需要在任务运行的时候，重设cbs的预算和截止时间，可以使用如下函数


rtems_cbs_set_parameters

至此，我们的测试程序如下


rtems_task Tasks_Periodic(
  rtems_task_argument argument
)
{
  rtems_id          rmid;
  rtems_id          test_rmid;
  rtems_status_code status;
  bool              scenario_done = 0;

  int start, stop, now;

  rtems_cbs_server_id server_id, tsid;
  rtems_cbs_parameters params, tparams;

  params.deadline = Periods[ argument ];
  params.budget = Execution[ argument ]+1;

  if ( argument == 4 ) {
    if ( rtems_cbs_create_server( &params, &overrun_handler_task_4, &server_id ))
      printf( "ERROR: CREATE SERVER FAILED\n" );
  }
  else {
    if ( rtems_cbs_create_server( &params, NULL, &server_id ) )
      printf( "ERROR: CREATE SERVER FAILED\n" );
  }
  if ( rtems_cbs_attach_thread( server_id, Task_id[ argument ] ) )
    printf( "ERROR: ATTACH THREAD FAILED\n" );
  if ( rtems_cbs_get_server_id( Task_id[ argument ], &tsid ) )
    printf( "ERROR: GET SERVER ID FAILED\n" );
  if ( tsid != server_id )
    printf( "ERROR: SERVER ID MISMATCH\n" );
  if ( rtems_cbs_get_parameters( server_id, &tparams ) )
    printf( "ERROR: GET PARAMETERS FAILED\n" );
  if ( params.deadline != tparams.deadline ||
       params.budget != tparams.budget )
    printf( "ERROR: PARAMETERS MISMATCH\n" );

  status = rtems_rate_monotonic_create( argument, &rmid );
  directive_failed( status, "rtems_rate_monotonic_create" );
  put_name( Task_name[ argument ], FALSE );
  printf( "- rtems_rate_monotonic_create id = 0x%08" PRIxrtems_id "\n",
          rmid );

  status = rtems_rate_monotonic_ident( argument, &test_rmid );
  directive_failed( status, "rtems_rate_monotonic_ident" );
  put_name( Task_name[ argument ], FALSE );
  printf( "- rtems_rate_monotonic_ident id = 0x%08" PRIxrtems_id "\n",
          test_rmid );

  if ( rmid != test_rmid ) {
     printf( "RMID's DO NOT MATCH (0x%" PRIxrtems_id " and 0x%"
             PRIxrtems_id ")\n", rmid, test_rmid );
     rtems_test_exit( 0 );
  }

  put_name( Task_name[ argument ], FALSE );
  printf( "- (0x%08" PRIxrtems_id ") period %" PRIu32 "\n",
          rmid, Periods[ argument ] );

  status = rtems_task_wake_after( 2 + Phases[argument] );
  directive_failed( status, "rtems_task_wake_after" );

  while (FOREVER) {
    if (rtems_rate_monotonic_period(rmid, Periods[argument])==RTEMS_TIMEOUT)
      printf("P%" PRIdPTR " - Deadline miss\n", argument);

    start = rtems_clock_get_ticks_since_boot();
    printf("P%" PRIdPTR "-S ticks:%d\n", argument, start);
    if ( start >= 2*HP_LENGTH ) break; /* stop */

    /* Specific scenario for task 4: tries to exceed announced budget,
       the task priority has to be pulled down to background. */
    now = rtems_clock_get_ticks_since_boot();
    if ( !scenario_done && argument == 4 && now >= 200 ) {
      Violating_task[ argument ] = 1;
      scenario_done = 1;
    }
    /* Specific scenario for task 3: changes scheduling parameters. */
    if ( !scenario_done && argument == 3 && now >= 250 ) {
      Periods[ argument ]   = Periods[ argument ] * 2;
      Execution[ argument ] = Execution[ argument ] * 2;
      params.deadline = Periods[ argument ];
      params.budget   = Execution[ argument ]+1;
      if ( rtems_cbs_set_parameters( server_id, &params) )
        printf( "ERROR: SET PARAMETERS FAILED\n" );
      if ( rtems_cbs_get_parameters( server_id, &tparams ) )
        printf( "ERROR: GET PARAMETERS FAILED\n" );
      if ( params.deadline != tparams.deadline ||
           params.budget != tparams.budget )
        printf( "ERROR: PARAMETERS MISMATCH\n" );
      scenario_done = 1;
    }
    /* Specific scenario for task 2: late unblock after being blocked by
       itself, the task priority has to be pulled down to background. */
    if ( !scenario_done && argument == 2 && now >= 500 ) {
      Violating_task[ argument ] = 1;
      scenario_done = 1;
    }
    if (argument == 2 && Violating_task[ argument ])
      rtems_task_wake_after( 10 );

    /* active computing */
    while(FOREVER) {
      now = rtems_clock_get_ticks_since_boot();
      if ( argument == 4 && !Violating_task[ argument ] &&
          (now >= start + Execution[argument]))
        break;
      if ( argument != 4 && (now >= start + Execution[argument]) )
        break;
    }
    stop = rtems_clock_get_ticks_since_boot();
    printf("P%" PRIdPTR "-F ticks:%d\n", argument, stop);
  }

  /* delete period and SELF */
  status = rtems_rate_monotonic_delete( rmid );
  if ( status != RTEMS_SUCCESSFUL ) {
    printf("rtems_rate_monotonic_delete failed with status of %d.\n",status);
    rtems_test_exit( 0 );
  }
  if ( rtems_cbs_cleanup() )
    printf( "ERROR: CBS CLEANUP\n" );
  fflush(stdout);
  TEST_END();
  rtems_test_exit( 0 );
}

代码比较容易理解，下面运行查看日志

日志


*** TEST CBS SCHEDULER 3 ***
PT1 - rtems_rate_monotonic_create id = 0x42010001
PT1 - rtems_rate_monotonic_ident id = 0x42010001
PT1 - (0x42010001) period 30
PT2 - rtems_rate_monotonic_create id = 0x42010002
PT2 - rtems_rate_monotonic_ident id = 0x42010002
PT2 - (0x42010002) period 40
PT3 - rtems_rate_monotonic_create id = 0x42010003
PT3 - rtems_rate_monotonic_ident id = 0x42010003
PT3 - (0x42010003) period 50
PT4 - rtems_rate_monotonic_create id = 0x42010004
PT4 - rtems_rate_monotonic_ident id = 0x42010004
PT4 - (0x42010004) period 70
AT5 AT6 P1-S ticks:2
P1-F ticks:12
P2-S ticks:12
P2-F ticks:22
P3-S ticks:22
P1-S ticks:32
P1-F ticks:42
P3-F ticks:42
P4-S ticks:42
P2-S ticks:52
P2-F ticks:62
P1-S ticks:62
P1-F ticks:72
P4-F ticks:72
P3-S ticks:72
P3-F ticks:82
AT6-S ticks:82
P6-F ticks:87
Killing task 6
AT5-S ticks:87
P1-S ticks:92
P1-F ticks:102
P2-S ticks:102
P2-F ticks:112
P4-S ticks:112
P1-S ticks:122
P1-F ticks:132
P3-S ticks:132
P3-F ticks:142
P2-S ticks:142
P2-F ticks:152
P4-F ticks:152
P1-S ticks:152
P1-F ticks:162
P2-S ticks:172
P2-F ticks:182
P1-S ticks:182
P1-F ticks:192
P3-S ticks:192
P3-F ticks:202
P4-S ticks:202
P1-S ticks:212
P1-F ticks:222
P4-F ticks:222
P2-S ticks:222
P2-F ticks:232
P3-S ticks:232
P3-F ticks:242
P1-S ticks:242
P1-F ticks:252
P2-S ticks:252
P2-F ticks:262
P4-S ticks:262
P1-S ticks:272
P1-F ticks:282
P3-S ticks:283
P3-F ticks:293
P2-S ticks:293
P2-F ticks:303
P1-S ticks:303
P1-F ticks:313
Signal overrun, fixing the task
P4-F ticks:313
P5-F ticks:313
Killing task 5
P3-S ticks:322
P1-S ticks:332
P1-F ticks:342
P3-F ticks:342
P2-S ticks:342
P2-F ticks:352
P4-S ticks:352
P4-F ticks:362
P1-S ticks:362
P1-F ticks:372
P2-S ticks:372
P2-F ticks:382
P3-S ticks:382
P1-S ticks:392
P1-F ticks:402
P4-S ticks:402
P2-S ticks:412
P2-F ticks:422
P1-S ticks:422
P1-F ticks:432
P4-F ticks:432
P3-F ticks:432
P1-S ticks:452
P1-F ticks:462
P2-S ticks:462
P2-F ticks:472
P4-S ticks:472
P1-S ticks:482
P1-F ticks:492
P4-F ticks:492
P2-S ticks:492
P2-F ticks:502
P3-S ticks:502
P1-S ticks:512
P1-F ticks:522
P3-F ticks:522
P2-S ticks:532
P4-S ticks:532
P1-S ticks:542
P1-F ticks:552
P4-F ticks:552
P2-F ticks:552
P1-S ticks:572
P1-F ticks:582
P2-S ticks:582
P3-S ticks:582
P1-S ticks:602
P1-F ticks:612
P3-F ticks:612
P4-S ticks:612
P4-F ticks:622
P2-F ticks:622
P2 - Deadline miss
P2-S ticks:622
P1-S ticks:632
P1-F ticks:642
P2-F ticks:642
P1-S ticks:662
P1-F ticks:672
P2-S ticks:672
P4-S ticks:672
P4-F ticks:682
P3-S ticks:682
P1-S ticks:692
P1-F ticks:702
P3-F ticks:702
P2-F ticks:702
P2 - Deadline miss
P2-S ticks:702
P2-F ticks:712
P1-S ticks:722
P1-F ticks:732
P2-S ticks:742
P4-S ticks:742
P1-S ticks:752
P1-F ticks:762
P4-F ticks:762
P2-F ticks:762
P3-S ticks:772
P1-S ticks:782
P1-F ticks:792
P2-S ticks:792
P3-F ticks:792
P2-F ticks:802
P1-S ticks:812
P1-F ticks:822
P2-S ticks:822
P4-S ticks:822
P4-F ticks:832
P2-F ticks:832
P1-S ticks:842
*** END OF TEST CBS SCHEDULER 3 ***

我们简单分析一下

首先，任务按照1>2>3>4的优先级运行，并且每个任务都会在自己周期内完成，我们可以留意0-82 ticks
然后，我们知道任务如果紧张，那么任务4在不能在周期内完成，那么overrun回调就会调用，例如任务4上次完成是222，下次完成在313时，花费了91ticks，故看到日志Signal overrun, fixing the task
最后，当任务在周期内，一直得不到优先级执行，那么当deadline达到的时候，在周期函数内会返回timeout，代表任务未在deadline中执行完毕，故看到从p2的ticks 582开始，到ticks 622之间，我们发现p2的任务一直被p3，p1，p4占用，导致最后p2在deadline上无法正常运行。

总结

至此，我们演示清楚了cbs调度器的示例，希望可以加深大家对cbs调度算法的理解

阅读全文

RTEMS调度器-恒定带宽(CBS)调度算法

编辑

2025-04-23

记录知识

根据之前的介绍，我们了解了edf调度算法，edf调度算法能够在理论上实现cpu利用率100%，也是单核性能最强的调度算法，因为edf算法基于deadline来控制调度的优先级，但是同样的，也是因为edf关于deadline的设计，我们使用这个算法的时候必须对任务的执行时间严格确定，也就是预设的周期内，任务是一定要完成的。如果deadline时间内，任务因为例如io问题导致延迟了，那么就会影响后面所有的任务的调度。从极端角度考虑，可能会导致deadline往后的任务，也就是优先级最低的任务，一直被饿死。
为了改善这个问题，cbs算法基于edf之上做了改进，本文介绍cbs算法

什么是cbs算法

cbs：constant bandwidth server。可以看到说明，cbs严格意义上只算一个服务，它用来改进edf算法，那么cbs支持哪些特性呢，如下：

为任务新增一个预算，此预算除以周期就是恒定带宽
如果任务的预算耗尽，但是任务没完成，则推迟deadline，降低任务优先级，让其他任务有机会运行
如果任务在周期内完成，则在下一个周期中，重新设置预算，重新运行，此预算就是其任务运行时间。

根据上面的描述，我们可以发现，cbs调度算法是一种更柔和的edf算法，如果我们的任务没办法确定固定的执行周期，也就是没办法保证任务在预设的周期内完成，那么edf算法就会像多米诺骨牌一样，一个任务延迟会导致其他所有的任务延迟。这种损坏是灾难性的。而cbs就不一样了，如果一个任务延迟，那么就会将其取消，然后降低其优先级。这样它就不会导致整个调度器上的就绪任务全部推迟。

rtems上的cbs调度算法

上面介绍清楚了cbs调度，它是edf的改进。接下来我们看一下rtems的cbs实现，如下


#define SCHEDULER_CBS_ENTRY_POINTS \
  { \
    _Scheduler_EDF_Initialize,       /* initialize entry point */ \
    _Scheduler_EDF_Schedule,         /* schedule entry point */ \
    _Scheduler_EDF_Yield,            /* yield entry point */ \
    _Scheduler_EDF_Block,            /* block entry point */ \
    _Scheduler_CBS_Unblock,          /* unblock entry point */ \
    _Scheduler_EDF_Update_priority,  /* update priority entry point */ \
    _Scheduler_EDF_Map_priority,     /* map priority entry point */ \
    _Scheduler_EDF_Unmap_priority,   /* unmap priority entry point */ \
    SCHEDULER_DEFAULT_SMP_OPERATIONS \
    _Scheduler_CBS_Node_initialize,  /* node initialize entry point */ \
    _Scheduler_default_Node_destroy, /* node destroy entry point */ \
    _Scheduler_CBS_Release_job,      /* new period of task */ \
    _Scheduler_CBS_Cancel_job,       /* cancel period of task */ \
    _Scheduler_default_Start_idle    /* start idle entry point */ \
    SCHEDULER_DEFAULT_SET_AFFINITY_OPERATION \
  }

可以发现，cbs上对任务的初始化，恢复阻塞，新增任务和取消任务做了改进。对于调度node的初始化，它新增了一个cbs的结构体，记录任务的deadline和budget，并提供一个当预算超时的函数回调。如下


typedef struct {
  /** Relative deadline of the server. */
  time_t deadline;
  /** Budget (computation time) of the server. */
  time_t budget;
} Scheduler_CBS_Parameters;
typedef struct {
  /**
   * Task id.
   *
   * @note: The current implementation of CBS handles only one task per server.
   */
  rtems_id                 task_id;
  /** Server paramenters. */
  Scheduler_CBS_Parameters parameters;
  /** Callback function invoked when a budget overrun occurs. */
  Scheduler_CBS_Budget_overrun  cbs_budget_overrun;

  /**
   * @brief Indicates if this CBS server is initialized.
   *
   * @see _Scheduler_CBS_Create_server() and _Scheduler_CBS_Destroy_server().
   */
  bool initialized;
} Scheduler_CBS_Server;

对于任务周期的开始，cbs相比于edf仅多做两件事

给任务节点设置一个预算
将任务的周期设置为deadline优先级节点

对于任务的取消，那么直接取消deadline的优先级节点即可

对于任务的解除阻塞，此实现比较重要，其函数如下


void _Scheduler_CBS_Unblock(
  const Scheduler_Control *scheduler,
  Thread_Control          *the_thread,
  Scheduler_Node          *node
)
{
  Scheduler_CBS_Node   *the_node;
  Scheduler_CBS_Server *serv_info;
  Priority_Control      priority;

  the_node = _Scheduler_CBS_Node_downcast( node );
  serv_info = the_node->cbs_server;
  priority = _Scheduler_Node_get_priority( &the_node->Base.Base );
  priority = SCHEDULER_PRIORITY_PURIFY( priority );

  /*
   * Late unblock rule for deadline-driven tasks. The remaining time to
   * deadline must be sufficient to serve the remaining computation time
   * without increased utilization of this task. It might cause a deadline
   * miss of another task.
   */
  if ( serv_info != NULL && ( priority & SCHEDULER_EDF_PRIO_MSB ) == 0 ) {
    time_t deadline = serv_info->parameters.deadline;
    time_t budget = serv_info->parameters.budget;
    uint32_t deadline_left = the_thread->CPU_budget.available;
    Priority_Control budget_left = priority - _Watchdog_Ticks_since_boot;

    if ( deadline * budget_left > budget * deadline_left ) {
      Thread_queue_Context queue_context;

      /* Put late unblocked task to background until the end of period. */
      _Thread_queue_Context_clear_priority_updates( &queue_context );
      _Scheduler_CBS_Cancel_job(
        scheduler,
        the_thread,
        the_node->deadline_node,
        &queue_context
      );
    }
  }

  _Scheduler_EDF_Unblock( scheduler, the_thread, &the_node->Base.Base );
}

可以看到，当任务从其他阻塞状态恢复到非阻塞状态，也就是加入就绪列表的时候，它会计算四个值

deadline
budget
剩余的deadline
剩余的budget

关于deadline和预算我们知道就是任务的周期和任务的执行事件，对于剩余的deadline和budget，其实就是任务实际运行时：

相较于设置的deadline还剩余的时间
相较于设置的budget还剩余的预算

那么关键的就是如下判断


if ( deadline * budget_left > budget * deadline_left ) {

我们看到不等式如下


deadline * budget_left > budget * deadline_left

它可以替换成如下


budget_left/deadline_left > budget/deadline

那么其含义就清楚了，那就是剩余的预算除以剩余的deadline的比重如果大于预期的预算除以deadline。那么就需要重设优先级，取消任务了。
再把意思转换一下，就是：

如果budget_left占比不变，则deadline_left占比太小，也就是剩余的deadline时间不够，也就是当前任务时间不够宽裕
如果deadline_left占比不变，则budget_left占比太大，也就是剩余的预算太大

对于上述的1好理解，如果deadline比重太小了，任务可能就不可控了。那么对于2怎么理解呢？

我们知道，如果任务正常执行，不可能存在剩余的预算除以剩余的deadline的占比反而大。但是在计算机领域，我们存在IO，IO是阻塞的，在IO期间，预算的时间是不被减掉的，也就是说进入unblock的预算时间值，实际上是进入IO之前的预算时间值。那么IO的时间实际上还没有计算进去。此时如果budget_left占比太大，说明任务有一定的风险导致不可控。这就理解清楚情况2了。

那么_Scheduler_CBS_Unblock的逻辑就是，如果剩余的任务占比要比预设的任务占比大，那么调度可能失控，此时将其任务从就绪队列删除，降低其优先级

总结

至此，我们基于edf介绍了cbs调度算法，它是edf算法的改进，当意外出现的时候，它会对失控的任务降低优先级，从而不会导致整个调度失控。
接下来，我会通过示例程序演示此调度器。

2025-04-21

根据《GICv3寄存器接口》的介绍，我们知道SPI/SGI/PPI 通过Distributor 找到

阅读全文