Max-min fairness

https://www.ece.rutgers.edu/~marsic/Teaching/CCN/minmax-fairsh.html

我们经常会遇到将稀缺资源分配给多个用户的情况，每个用户获得资源的权利都相同，但是需求数却可能不同，这个时候应该怎么办？一种常用的方法叫 max-min fairness share，尽量满足用户中的最小的需求，然后将剩余的资源公平分配给剩下的用户。形式化定义如下：

1. 资源分配以需求递增的方式进行分配
2. 每个用户获得的资源不超过其需求
3. 未得到满足的用户等价平分剩下的资源

更形象的表达如下，有用户 1, …, n，资源需求分别为 x₁ , x₂ , …, x_n .不失一般性，假设 x₁ <= x₂ <= … <= x_n 。假设资源总数为 C。然后先给需求最小 x₁ 的用户分配资源数 C/n 。如果 x₁ < C/n，则用户 1 剩余资源 C/n - x₁，则剩下的用户每人获取资源数为 C/n + (C/n - x₁)/(n-1)，然后依次处理剩下的用户，直到资源数小于等于需求数或者所有用户分配完。举个例子如下：

用户 1, 2, 3, 4 的需求分别为 2, 2.6, 4, 5，总资源数为 10。首先我们将资源数平分 4 等份，10/4 = 2.5。用户 1 需求是 2，则剩余 2.5 - 2 = 0.5，0.5 再平分给用户 2，3，4，0.5/3 = 0.1666…。然后用户 2 需求 2.6，剩余 2.5 + 0.1666 - 2.6 = 0.066…，再平分，0.066/2 = 0.033…，则用户 3 获得 2.5 + 0.66… + 0.033… = 2.7 ，此时不满足资源需求，结束。最终 4 个用户获得的资源数依次为 2, 2.6, 2.7, 2.7。

Weighted max-min fairness

上面我们假设每个用户获得资源的权重一样，对于权重不一样的处理方式也就是 weighted max-min fairness share。用户 1,2,…,n 的权重分别为 w₁, w₂ , …, w_n ，这也就意味我们第一次总资源切分以及之后多余资源切分不是等分，而是按权重来切分。举例如下：

用户 1, 2, 3, 4 需要的资源数分别为 4, 2, 10, 4，权重分别为 2.5, 4, 0.5, 1，资源总数为 16。首先对权重做处理为 5, 8, 1, 2，然后将资源 16 按权重切分每个用户获得资源数：5, 8, 1, 2。我们先遍历可以满足需求的用户：用户 1 需要资源 4，得到 5， 剩余 1。用户需要资源 2，得到 8，剩余 6。用户 3 和 4 资源无法满足。总剩余资源数为 7，按权重分给用户 3 和 4，用户 4 得到资源 7 _ 2/3 + 2 = 6.666 多于需求 4，剩下 2.666，分给用户 3。最后用户 3 获得资源数为 1 + 7 _ 1/3 + 2.666 = 6。

特殊情况

如果任何用户均无法被满足，则相当于平均分配。

IMCI Auto DOP 具体实现

定义用户

将并发状态下被允许执行的查询称为用户，举个例子，如果系统最大 dop 为 8， 有 10000 个连接，每个连接都在不停地发送请求，那么现在的 QueryQueue 会按照 FCFS 的方法选择 8 个连接发送的查询开始执行，其他请求排在队列中，当这个 8 个中的一个执行完毕，再按照 FCFS 的方法选择下一个。那么每次选择的 8 个查询，即对应 8 个用户。

定义资源需求

每个用户的资源需求在没有特殊要求的情况下均定义为系统最大 dop(假设为 8)，但对于以下情况特殊处理：

1. manual dop 这种用户不受到公平策略影响，始终满足 manual dop 的要求，manual dop 会影响系统的并发程度，如果 manual dop = 8, 则需要等到之前的查询全部结束，并阻塞后面的全部用户。manual dop 设置为大于 8 的情况应当被禁止。
2. max dop，先按照普通用户处理，如果发现处理结果超过 max dop，则减少到 max dop.

举一个例子：

例如 4 个查询并发，其要求如下：

下面通过具体代码来看这个结果如何达到

int64_t concurrency = stats_.Concurrency(); // 4 in this example
int64_t avg_dop = max_dop_one_query_; // 8

if (concurrency > 1) {
    avg_dop = max_dop_ / concurrency; // 2
}

首先按照没有特殊情况计算 avg_dop, 使用 max-min fairness 的特殊情况，即认为所有用户的需求都无法被满足，此时平分 dop，因此 avg dop 为 2

// Step 1: Get real concurrency
int64_t dop_left = max_dop_ - stats_.DopRunning();
bool memory_bound = false;
for (Query *query : queue_) {
    int64_t choosed_dop = avg_dop;
    if (query->manual_dop_ != 0) {
        choosed_dop = query->manual_dop_;
    } else if (choosed_dop > query->max_dop_) {
        choosed_dop = query->max_dop_;
    }

    if (dop_left < choosed_dop) {
        break;
    }
    if (!query->TryAcquireMemoryQuota()) {
        memory_bound = true;
        break;
    }
    dop_left -= choosed_dop;
    query->dop_allocated_ = choosed_dop;
    ++scheduled_count;
}

系统按照先到先得的顺序遍历队列中的查询，在这里处理 manual dop 和 max dop,

1. 第一个请求，manual dop = 4，_ query->dop_allocated_ _= choosed_dop = 4，dop_left = 4
2. 第二个请求，max dop = 1, query->dop_allocated__ _= choosed_dop = 1, dop_left = 3
3. 第三个请求，没有特殊要求，query->dop_allocated__ _= choosed_dop = avg_dop = 2, dop_left= 1
4. choosed_dop = avg_dop = 2, dop_left < choosed_dop, 本轮调度结束

与在 Max-min fairess 中不同的是，第三步，3 个 dop 将被平分到第三个请求和第四个请求，即每个请求获得 1.5 个 dop，但由于 dop 的离散特性，无法这样做，依然使用 avg_dop。这样做的缺陷是如果 manual dop 和 max dop 占比较多，那么对 avg dop 会产生影响，因此需要一个反馈链路来优化 avg_dop。目前也可以将 dop 设置为 core 数量的 2 倍解决，例如 16，此时总计分配出的 dop 数量大概率在 8～16，这样可以依赖 OS 实现公平调度。

内存

内存是一个需要额外考虑的问题，max-min fairness 是针对单一资源的，当考虑到内存、CPU、IO 资源时 DRF 可能会更好，但是暂时没有想到如何将它利用起来。

将内存限制视为对并发数的修改，在此基础上重新分配剩余的 DOP。在上面一段代码中，如果 break 的原因是内存 quota 已耗尽，memory bound 的标记会被设置，此时 dop_left 会按照新的并发数量被重新分配给已被允许执行的查询。

举个例子：系统 core 数量 8，最大 dop 16，内存 32 GB

按照上一段代码 step 1 的流程，系统会按上表中的流程执行，最终内存耗尽，memory_bound = true, 剩余 dop 为 8. 接下来

 // Step 2: Conpensate DOP if the system is memory-bound
  int32_t scheduled_left = scheduled_count;
  concurrency = stats_.QueryRunning() + scheduled_left;
  if (scheduled_left > 0) {  // scheduled_left means concurrency > 0
    avg_dop = std::min(max_dop_ / concurrency, dop_left / scheduled_left);
  }
  for (Query *query : queue_) {
    if (scheduled_left <= 0) {
      break;
    }

    if (memory_bound && 0 == query->hint_dop_ &&
        query->dop_allocated_ < query->max_dop_) {
      query->dop_allocated_ = std::min(query->max_dop_, avg_dop);
    }

    query->StartToRun(query->dop_allocated_);
    stats_.UpdateForSchedule(query->dop_allocated_);
    IMCI_EXECUTION_LOG_DEBUG("[AUTODOP]", K(query->dop_allocated_), K(avg_dop),
                             K(dop_left), K(scheduled_left), K(concurrency));
    --scheduled_left;
  }

先按照 memory bound 重新计算 concurrency, avg_dop，

avg_dop = std::min(max_dop_ / concurrency, dop_left / scheduled_left); 含义是当前 dop 不够平均时，就少用一些，这样下一次调度时会有更多的空闲 dop，此时结果会逐步趋近 max_dop_ / concurrency. 这里计算得到新的 avg_dop 是 16 / 4 = 4, 最终分配结果为 4, 3, 3, 4 剩余两个线程，依靠 OS 实现公平。