限流算法：令牌桶 / 漏桶 / GCRA

#include <time.h>
#include <stdint.h>
#include <stdbool.h>

typedef struct {
    int64_t  window_start;   /* 当前窗口起始时间（秒） */
    int64_t  count;          /* 当前窗口内的请求计数 */
    int64_t  limit;          /* 窗口内允许的最大请求数 */
    int64_t  window_size;    /* 窗口大小（秒） */
} fixed_window_t;

bool fixed_window_allow(fixed_window_t *fw) {
    int64_t now = time(NULL);
    int64_t window = now / fw->window_size * fw->window_size;

    if (window != fw->window_start) {
        fw->window_start = window;
        fw->count = 0;
    }

    if (fw->count < fw->limit) {
        fw->count++;
        return true;
    }
    return false;
}

#include <stdlib.h>
#include <time.h>
#include <stdbool.h>

typedef struct {
    double  *timestamps;   /* 环形缓冲区，存储请求时间戳 */
    int      capacity;     /* 缓冲区容量 */
    int      head;         /* 头指针 */
    int      tail;         /* 尾指针 */
    int      count;        /* 当前请求数 */
    int      limit;        /* 窗口内允许的最大请求数 */
    double   window_size;  /* 窗口大小（秒） */
} sliding_log_t;

static double now_sec(void) {
    struct timespec ts;
    clock_gettime(CLOCK_MONOTONIC, &ts);
    return ts.tv_sec + ts.tv_nsec * 1e-9;
}

/* 清理过期的时间戳 */
static void evict_expired(sliding_log_t *sl, double now) {
    double cutoff = now - sl->window_size;
    while (sl->count > 0 && sl->timestamps[sl->head] <= cutoff) {
        sl->head = (sl->head + 1) % sl->capacity;
        sl->count--;
    }
}

bool sliding_log_allow(sliding_log_t *sl) {
    double now = now_sec();
    evict_expired(sl, now);

    if (sl->count < sl->limit) {
        sl->timestamps[sl->tail] = now;
        sl->tail = (sl->tail + 1) % sl->capacity;
        sl->count++;
        return true;
    }
    return false;
}

typedef struct {
    int64_t  prev_window;    /* 上一个窗口的起始时间 */
    int64_t  prev_count;     /* 上一个窗口的请求计数 */
    int64_t  curr_window;    /* 当前窗口的起始时间 */
    int64_t  curr_count;     /* 当前窗口的请求计数 */
    int64_t  limit;          /* 窗口内允许的最大请求数 */
    int64_t  window_size;    /* 窗口大小（秒） */
} sliding_counter_t;

bool sliding_counter_allow(sliding_counter_t *sc) {
    int64_t now = time(NULL);
    int64_t curr_win = now / sc->window_size * sc->window_size;

    if (curr_win != sc->curr_window) {
        sc->prev_count  = (curr_win - sc->curr_window == sc->window_size)
                          ? sc->curr_count : 0;
        sc->prev_window = sc->curr_window;
        sc->curr_window = curr_win;
        sc->curr_count  = 0;
    }

    /* 计算当前时刻在窗口内的位置比例 */
    double elapsed = (double)(now - curr_win) / sc->window_size;
    /* 加权估计：上一个窗口的"剩余"部分 + 当前窗口的计数 */
    double estimated = sc->prev_count * (1.0 - elapsed) + sc->curr_count;

    if (estimated < sc->limit) {
        sc->curr_count++;
        return true;
    }
    return false;
}

#include <stdbool.h>
#include <time.h>

typedef struct {
    double   water;          /* 桶中当前水量 */
    double   capacity;       /* 桶的容量（最大排队数） */
    double   leak_rate;      /* 漏出速率（请求/秒） */
    double   last_leak_time; /* 上次漏水时间 */
} leaky_bucket_t;

static double monotonic_sec(void) {
    struct timespec ts;
    clock_gettime(CLOCK_MONOTONIC, &ts);
    return ts.tv_sec + ts.tv_nsec * 1e-9;
}

static void leak(leaky_bucket_t *lb) {
    double now = monotonic_sec();
    double elapsed = now - lb->last_leak_time;
    double leaked = elapsed * lb->leak_rate;

    lb->water -= leaked;
    if (lb->water < 0) lb->water = 0;
    lb->last_leak_time = now;
}

bool leaky_bucket_allow(leaky_bucket_t *lb) {
    leak(lb);

    if (lb->water < lb->capacity) {
        lb->water += 1.0;
        return true;
    }
    return false;   /* 桶满，拒绝 */
}

typedef struct {
    double   tokens;         /* 当前令牌数 */
    double   capacity;       /* 桶容量（最大令牌数） */
    double   rate;           /* 令牌产生速率（tokens/sec） */
    double   last_refill;    /* 上次补充时间 */
} token_bucket_t;

static void refill(token_bucket_t *tb) {
    double now = monotonic_sec();
    double elapsed = now - tb->last_refill;
    double new_tokens = elapsed * tb->rate;

    tb->tokens += new_tokens;
    if (tb->tokens > tb->capacity)
        tb->tokens = tb->capacity;
    tb->last_refill = now;
}

bool token_bucket_allow(token_bucket_t *tb, double cost) {
    refill(tb);

    if (tb->tokens >= cost) {
        tb->tokens -= cost;
        return true;
    }
    return false;
}

typedef struct {
    token_bucket_t  cir_bucket;   /* CIR 令牌桶 */
    token_bucket_t  pir_bucket;   /* PIR 令牌桶 */
} dual_rate_tb_t;

typedef enum { GREEN, YELLOW, RED } color_t;

color_t dual_rate_classify(dual_rate_tb_t *dr, double cost) {
    refill(&dr->pir_bucket);
    refill(&dr->cir_bucket);

    /* PIR 桶没令牌 -> 红色（丢弃） */
    if (dr->pir_bucket.tokens < cost)
        return RED;

    /* CIR 桶没令牌 -> 黄色（尽力转发） */
    if (dr->cir_bucket.tokens < cost) {
        dr->pir_bucket.tokens -= cost;
        return YELLOW;
    }

    /* 两个桶都有令牌 -> 绿色（保证转发） */
    dr->cir_bucket.tokens -= cost;
    dr->pir_bucket.tokens -= cost;
    return GREEN;
}

typedef struct {
    double   tat;             /* Theoretical Arrival Time */
    double   emission_interval; /* T = 1/rate */
    double   burst_tolerance;   /* tau = T * (burst - 1) */
} gcra_t;

void gcra_init(gcra_t *g, double rate, int burst) {
    g->tat = 0;
    g->emission_interval = 1.0 / rate;
    g->burst_tolerance = g->emission_interval * (burst - 1);
}

bool gcra_allow(gcra_t *g) {
    double now = monotonic_sec();

    /* 新的 TAT：取 TAT 和当前时间的较大值，加上发送间隔 */
    double new_tat = (g->tat > now ? g->tat : now) + g->emission_interval;

    /* 如果新的 TAT 超过当前时间 + 突发容忍度，则拒绝 */
    if (new_tat - now > g->emission_interval + g->burst_tolerance) {
        return false;
    }

    g->tat = new_tat;
    return true;
}

# 错误示范：存在竞态条件
tokens = redis.get("rate:user:123")
if tokens > 0:
    redis.set("rate:user:123", tokens - 1)  # TOCTOU 漏洞
    allow()
else:
    reject()

-- token_bucket.lua
-- KEYS[1] = rate limiter key
-- ARGV[1] = capacity (burst size)
-- ARGV[2] = rate (tokens per second)
-- ARGV[3] = now (current timestamp in microseconds)
-- ARGV[4] = cost (tokens to consume, default 1)

local key       = KEYS[1]
local capacity  = tonumber(ARGV[1])
local rate      = tonumber(ARGV[2])
local now       = tonumber(ARGV[3])
local cost      = tonumber(ARGV[4]) or 1

-- 获取当前状态
local data = redis.call('HMGET', key, 'tokens', 'last_refill')
local tokens      = tonumber(data[1]) or capacity
local last_refill = tonumber(data[2]) or now

-- 补充令牌
local elapsed    = math.max(0, now - last_refill) / 1000000.0
local new_tokens = math.min(capacity, tokens + elapsed * rate)

-- 判断是否允许
local allowed = 0
local new_cost = cost
if new_tokens >= cost then
    new_tokens = new_tokens - cost
    allowed = 1
end

-- 更新状态
redis.call('HMSET', key,
    'tokens', tostring(new_tokens),
    'last_refill', tostring(now))

-- 设置过期时间，防止无限占用内存
-- 过期时间 = 桶填满所需时间 * 2
local ttl = math.ceil(capacity / rate * 2)
redis.call('EXPIRE', key, ttl)

-- 返回：[是否允许, 剩余令牌数]
return {allowed, tostring(new_tokens)}

-- gcra.lua
-- KEYS[1] = rate limiter key
-- ARGV[1] = emission_interval (microseconds)
-- ARGV[2] = burst_tolerance (microseconds)
-- ARGV[3] = now (current timestamp in microseconds)

local key               = KEYS[1]
local emission_interval = tonumber(ARGV[1])
local burst_tolerance   = tonumber(ARGV[2])
local now               = tonumber(ARGV[3])

local tat = tonumber(redis.call('GET', key)) or now

-- new_tat = max(tat, now) + emission_interval
local new_tat = math.max(tat, now) + emission_interval

-- 允许条件: new_tat - now <= emission_interval + burst_tolerance
local allow_at = now - burst_tolerance
local diff = new_tat - now

if diff > emission_interval + burst_tolerance then
    -- 限流：计算需要等待的时间
    local retry_after = diff - emission_interval - burst_tolerance
    return {0, tostring(retry_after)}
end

-- 允许：更新 TAT
redis.call('SET', key, tostring(new_tat))

-- TTL: 突发窗口 + 发送间隔，转换为秒并向上取整
local ttl = math.ceil((emission_interval + burst_tolerance) / 1000000)
redis.call('EXPIRE', key, ttl + 1)

-- 返回: [是否允许, 剩余突发容量（微秒）]
local remaining = emission_interval + burst_tolerance - diff
return {1, tostring(remaining)}

/* token_bucket.c -- 完整的令牌桶实现 */
#include <stdio.h>
#include <stdlib.h>
#include <stdbool.h>
#include <string.h>
#include <time.h>
#include <math.h>
#include <pthread.h>
#include <errno.h>

/* ========== 时间工具 ========== */

static double clock_now(void) {
    struct timespec ts;
    clock_gettime(CLOCK_MONOTONIC, &ts);
    return (double)ts.tv_sec + (double)ts.tv_nsec * 1e-9;
}

/* ========== 令牌桶结构 ========== */

typedef struct token_bucket {
    double           tokens;       /* 当前令牌数 */
    double           capacity;     /* 桶容量（最大令牌数） */
    double           rate;         /* 令牌产生速率（tokens/sec） */
    double           last_refill;  /* 上次补充时间戳 */
    pthread_mutex_t  lock;         /* 线程安全锁 */
} token_bucket_t;

/* 初始化令牌桶 */
int tb_init(token_bucket_t *tb, double rate, double capacity) {
    if (rate <= 0 || capacity <= 0) {
        errno = EINVAL;
        return -1;
    }

    tb->tokens      = capacity;   /* 初始满桶 */
    tb->capacity    = capacity;
    tb->rate        = rate;
    tb->last_refill = clock_now();

    if (pthread_mutex_init(&tb->lock, NULL) != 0)
        return -1;

    return 0;
}

/* 销毁令牌桶 */
void tb_destroy(token_bucket_t *tb) {
    pthread_mutex_destroy(&tb->lock);
}

/* 内部：补充令牌（调用者持有锁） */
static void tb_refill_locked(token_bucket_t *tb) {
    double now     = clock_now();
    double elapsed = now - tb->last_refill;

    if (elapsed > 0) {
        tb->tokens += elapsed * tb->rate;
        if (tb->tokens > tb->capacity)
            tb->tokens = tb->capacity;
        tb->last_refill = now;
    }
}

/* 尝试消耗令牌 */
bool tb_consume(token_bucket_t *tb, double cost) {
    if (cost <= 0) return true;

    pthread_mutex_lock(&tb->lock);
    tb_refill_locked(tb);

    bool allowed = false;
    if (tb->tokens >= cost) {
        tb->tokens -= cost;
        allowed = true;
    }

    pthread_mutex_unlock(&tb->lock);
    return allowed;
}

/* 查询当前令牌数（不消耗） */
double tb_tokens(token_bucket_t *tb) {
    pthread_mutex_lock(&tb->lock);
    tb_refill_locked(tb);
    double t = tb->tokens;
    pthread_mutex_unlock(&tb->lock);
    return t;
}

/* 查询需要等待多久才能获得指定数量的令牌 */
double tb_wait_time(token_bucket_t *tb, double cost) {
    pthread_mutex_lock(&tb->lock);
    tb_refill_locked(tb);

    double wait = 0;
    if (tb->tokens < cost) {
        double deficit = cost - tb->tokens;
        wait = deficit / tb->rate;
    }

    pthread_mutex_unlock(&tb->lock);
    return wait;
}

/* 阻塞等待直到获得令牌 */
bool tb_consume_wait(token_bucket_t *tb, double cost, double timeout_sec) {
    double deadline = clock_now() + timeout_sec;

    while (1) {
        if (tb_consume(tb, cost))
            return true;

        double wait = tb_wait_time(tb, cost);
        double remaining = deadline - clock_now();

        if (remaining <= 0)
            return false;  /* 超时 */

        if (wait > remaining)
            wait = remaining;

        /* 精确等待 */
        struct timespec ts;
        ts.tv_sec  = (time_t)wait;
        ts.tv_nsec = (long)((wait - ts.tv_sec) * 1e9);
        nanosleep(&ts, NULL);
    }
}

/* ========== 测试 ========== */

int main(void) {
    token_bucket_t tb;
    int i, allowed;

    /* 基础测试：10 tokens/sec, burst 5 */
    tb_init(&tb, 10.0, 5.0);
    printf("=== 基础测试 ===\n");
    for (i = 0; i < 6; i++) {
        bool ok = tb_consume(&tb, 1.0);
        printf("  请求 %d: %s (剩余 %.1f)\n",
               i + 1, ok ? "允许" : "拒绝", tb_tokens(&tb));
    }
    struct timespec w1 = {0, 300000000};  /* 等待 0.3s */
    nanosleep(&w1, NULL);
    printf("  等待 0.3s 后，令牌数: %.1f\n", tb_tokens(&tb));
    tb_destroy(&tb);

    /* 突发测试：100 tokens/sec, burst 20 */
    tb_init(&tb, 100.0, 20.0);
    printf("\n=== 突发测试 ===\n");
    struct timespec w2 = {1, 0};
    nanosleep(&w2, NULL);
    for (i = 0, allowed = 0; i < 30; i++)
        if (tb_consume(&tb, 1.0)) allowed++;
    printf("  发送 30 请求，允许 %d 个（预期约 20）\n", allowed);
    tb_destroy(&tb);

    /* 可变消耗测试 */
    tb_init(&tb, 50.0, 10.0);
    printf("\n=== 可变消耗测试 ===\n");
    printf("  消耗 3: %s\n", tb_consume(&tb, 3.0) ? "允许" : "拒绝");
    printf("  消耗 5: %s\n", tb_consume(&tb, 5.0) ? "允许" : "拒绝");
    printf("  消耗 5: %s (预期拒绝)\n", tb_consume(&tb, 5.0) ? "允许" : "拒绝");
    tb_destroy(&tb);

    printf("\n所有测试完成。\n");
    return 0;
}

/* gcra.c -- 完整的 GCRA 实现 */
#include <stdio.h>
#include <stdlib.h>
#include <stdbool.h>
#include <string.h>
#include <time.h>
#include <math.h>
#include <pthread.h>
#include <errno.h>

/* ========== 时间工具 ========== */

static double clock_now(void) {
    struct timespec ts;
    clock_gettime(CLOCK_MONOTONIC, &ts);
    return (double)ts.tv_sec + (double)ts.tv_nsec * 1e-9;
}

/* ========== GCRA 结构 ========== */

typedef struct gcra {
    double           tat;                /* Theoretical Arrival Time */
    double           emission_interval;  /* T = 1/rate */
    double           burst_tolerance;    /* tau = T * (burst - 1) */
    double           rate;               /* 请求速率 */
    int              burst;              /* 突发容量 */
    pthread_mutex_t  lock;               /* 线程安全锁 */
} gcra_t;

/* 初始化 GCRA */
int gcra_init(gcra_t *g, double rate, int burst) {
    if (rate <= 0 || burst <= 0) {
        errno = EINVAL;
        return -1;
    }

    g->emission_interval = 1.0 / rate;
    g->burst_tolerance   = g->emission_interval * (burst - 1);
    g->tat               = 0;   /* 初始 TAT 为 0，表示空闲 */
    g->rate              = rate;
    g->burst             = burst;

    if (pthread_mutex_init(&g->lock, NULL) != 0)
        return -1;

    return 0;
}

/* 销毁 GCRA */
void gcra_destroy(gcra_t *g) {
    pthread_mutex_destroy(&g->lock);
}

/* 请求结果 */
typedef struct {
    bool    allowed;       /* 是否允许 */
    double  retry_after;   /* 如果拒绝，需要等待的秒数 */
    double  remaining;     /* 剩余突发容量（等价令牌数） */
    double  reset_after;   /* TAT 回到当前时间之前需要的秒数 */
} gcra_result_t;

/* 判断请求是否允许 */
gcra_result_t gcra_check(gcra_t *g) {
    gcra_result_t result;
    double now = clock_now();

    pthread_mutex_lock(&g->lock);

    /* 如果 TAT 落后于当前时间，说明系统空闲 */
    double tat_or_now = g->tat > now ? g->tat : now;
    double new_tat = tat_or_now + g->emission_interval;
    double diff = new_tat - now;
    double limit = g->emission_interval + g->burst_tolerance;

    if (diff > limit) {
        /* 拒绝：TAT 太超前 */
        result.allowed     = false;
        result.retry_after = diff - limit;
        result.remaining   = 0;
        result.reset_after = g->tat - now;
    } else {
        /* 允许：更新 TAT */
        g->tat = new_tat;
        result.allowed     = true;
        result.retry_after = 0;
        result.remaining   = (limit - diff) / g->emission_interval;
        result.reset_after = new_tat - now;
    }

    pthread_mutex_unlock(&g->lock);
    return result;
}

/* 简单的允许/拒绝判断 */
bool gcra_allow(gcra_t *g) {
    gcra_result_t r = gcra_check(g);
    return r.allowed;
}

/* 阻塞等待直到请求被允许 */
bool gcra_allow_wait(gcra_t *g, double timeout_sec) {
    double deadline = clock_now() + timeout_sec;

    while (1) {
        gcra_result_t r = gcra_check(g);
        if (r.allowed)
            return true;

        if (clock_now() + r.retry_after > deadline)
            return false;  /* 超时 */

        struct timespec ts;
        ts.tv_sec  = (time_t)r.retry_after;
        ts.tv_nsec = (long)((r.retry_after - ts.tv_sec) * 1e9);
        nanosleep(&ts, NULL);
    }
}

/* 查询当前状态（不消耗） */
double gcra_remaining(gcra_t *g) {
    double now = clock_now();
    pthread_mutex_lock(&g->lock);

    double tat_or_now = g->tat > now ? g->tat : now;
    double diff = tat_or_now + g->emission_interval - now;
    double limit = g->emission_interval + g->burst_tolerance;
    double remaining = (limit - diff) / g->emission_interval;

    if (remaining < 0) remaining = 0;
    pthread_mutex_unlock(&g->lock);
    return remaining;
}

/* ========== 测试 ========== */

int main(void) {
    gcra_t g;
    int i, allowed;

    /* 基础测试：10 req/s, burst 5 */
    gcra_init(&g, 10.0, 5);
    printf("=== GCRA 基础测试 ===\n");
    for (i = 0; i < 7; i++) {
        gcra_result_t r = gcra_check(&g);
        printf("  请求 %d: %s (剩余 %.1f)\n",
               i + 1, r.allowed ? "允许" : "拒绝", r.remaining);
    }
    gcra_destroy(&g);

    /* 恢复测试 */
    gcra_init(&g, 10.0, 5);
    printf("\n=== GCRA 恢复测试 ===\n");
    for (i = 0; i < 5; i++) gcra_allow(&g);
    printf("  消耗 5 个后: 剩余 %.1f\n", gcra_remaining(&g));
    struct timespec w = {0, 300000000};
    nanosleep(&w, NULL);
    printf("  等待 0.3s 后: 剩余 %.1f\n", gcra_remaining(&g));
    for (i = 0, allowed = 0; i < 5; i++)
        if (gcra_allow(&g)) allowed++;
    printf("  再发 5 个，允许 %d 个（预期约 3）\n", allowed);
    gcra_destroy(&g);

    /* 等价性测试 */
    gcra_init(&g, 20.0, 10);
    printf("\n=== 等价性测试 ===\n");
    int total = 0; allowed = 0;
    double start = clock_now();
    while (clock_now() - start < 1.0) {
        total++;
        if (gcra_allow(&g)) allowed++;
        struct timespec d = {0, 10000000};
        nanosleep(&d, NULL);
    }
    printf("  1s 内发送 %d，允许 %d（预期约 30）\n", total, allowed);
    gcra_destroy(&g);

    printf("\n所有 GCRA 测试完成。\n");
    return 0;
}

# 编译令牌桶
gcc -O2 -Wall -o token_bucket token_bucket.c -lpthread -lm

# 编译 GCRA
gcc -O2 -Wall -o gcra gcra.c -lpthread -lm

# 运行测试
./token_bucket
./gcra

# 创建 HTB 根队列，总带宽 100Mbps
tc qdisc add dev eth0 root handle 1: htb default 30
tc class add dev eth0 parent 1: classid 1:1 htb rate 100mbit burst 15k

# 企业 A：保证 40Mbps，最高 80Mbps
tc class add dev eth0 parent 1:1 classid 1:10 htb \
    rate 40mbit ceil 80mbit burst 15k

# 企业 A 下用户 1：保证 10Mbps，最高 30Mbps
tc class add dev eth0 parent 1:10 classid 1:11 htb \
    rate 10mbit ceil 30mbit burst 10k

# 过滤器：将流量分配到对应的类
tc filter add dev eth0 parent 1: protocol ip prio 1 \
    u32 match ip src 10.0.1.0/24 flowid 1:11

http_filters:
- name: envoy.filters.http.ratelimit
  typed_config:
    "@type": type.googleapis.com/envoy.extensions.filters.http.ratelimit.v3.RateLimit
    domain: production
    rate_limit_service:
      grpc_service:
        envoy_grpc:
          cluster_name: rate_limit_service