System.Threading.RateLimitingで流量制限を行う

始めに

dotnet-7から、System.Threading.Ratelimitingパッケージが追加された。
この記事では、クラス構造や基本的な使い方を解説する。
実装は https://github.com/dotnet/runtime/tree/main/src/libraries/System.Threading.RateLimiting を参照。
また、APIの提案内容は https://github.com/dotnet/runtime/issues/65400 に記載されている。

何をするためのものか

例えばウェブアプリ等で、送受信データ量等を一定の水準で抑制したいときに使う。
また、バッチ処理等でも、一度に処理する量に対して制限をかけ、CPU負荷を抑制したいときに使えるのではないかと思う。
標準ではインプロセスかつオンメモリの実装しかないが、インターフェイスだけ使って他の実装を作る事も可能。

個人的には YARP のために追加されたものではないかと思っている。
この機能があれば、IISやnginx等のHTTPサーバーの機能に頼ることなく、C#のみできめ細かい制御が可能となる。

プロジェクトへの導入

TargetFrameworkをnet7.0以降にすれば、特にパッケージ追加無しで使うことは可能。
それ以外で使いたい場合は、 System.Threading.RateLimitingのnugetパッケージ をプロジェクト参照に追加する。
netstandard2.0もカバーしてるので、net4xなプロジェクトでも利用可能

プロジェクト参照が追加されれば、System.Threading.RateLimiting以下の各種クラスが使用可能になる。

クラス構造

ベースクラス

System.Threading.RateLimiting以下で、最低限使用するクラスは以下で、全て仮想クラス。

• RateLimiter
• RateLimitLease

最も単純な流れとしては、

1. ユーザーがRateLimiterを継承する実装のインスタンスを作成する
   • 各実装については後述する
   • 大体はシングルトン
2. RateLimitLease RateLimiter.AttemptAcquire(int permitCount = 1) あるいは ValueTask<RateLimitLease> AcquireAsync(int permitCount = 1, CancellationToken token = default) で、リソースの占有を試みる
   • permitCountは、どの程度のリソースを占有するかという要求数
     • ネットワーク流量制限であればバイト数等
     • permitCount=0の時、RateLimiterは現在空きがあるかどうかを確認するだけという特殊動作を行うことが推奨されている
   • AttemptAcquireはキューに入らずにすぐに結果が返ってくるが、AcquireAsyncは要求をキューに入れられた上で結果を待たされる
3. bool RateLimitLease.IsAcquired で、占有が成功したか確認する
4. 占有を開放する時は、RateLimitLeaseをDisposeする
   • 実装によっては必要ないものもあるが、しておいた方が無難

実装にもよるが、待たされた挙句確保失敗というのは十分あり得ることなので、ユーザーはそれも織り込んで使用する必要がある。

RateLimiter

制限を行うための基点となる。ユーザーはこのクラスを通してリソースの占有等を行う。
仮想クラスとなっており、最低限実装するべきは以下のものとなる。

• protected abstract RateLimitLease AttemptAcquireCore(int permitCount)
• protected abstract ValueTask<RateLimitLease> AcquireAsyncCore(int permitCount, CancellationToken cancellationToken = default)
• public abstract int GetAvailablePermits()
• public abstract TimeSpan? IdleDuration { get; }
• public abstract RateLimiterStatistics GetStatistics()

他は、後処理のために protected void Dispose(bool disposing) と ValueTask DisposeAsyncCore() をオーバーライドする必要があるかもしれない。

RateLimitLease

RateLimiterが、取得の結果として返すことになるクラス。
ロックが成功したかどうか、あるいはRateLimiterの実装クラスが追加したメタデータ等を取得することができる。
これをDisposeすることで、リソースの解放を示すことができる。
後述するTokenBucket方式のように必ずしも解放をする必要がないものもあるが、基本的にはDisposeはしておいた方が良いだろう。
最低限実装すべきものは以下のものとなる。

• public abstract bool IsAcquired { get; }
• public abstract bool TryGetMetadata(string metadataName, out object? metadata)
• public abstract IEnumerable<string> MetadataNames { get; }

後は、リソースの解放をRateLimiterに知らせるために、 protected virtual void Dispose(bool disposing) をオーバーライドすることになると思う。DisposeAsyncは継承してないので注意。

RateLimiterStatistics

現在のRateLimiterの統計情報を表すクラス。
下記の { get; init; } なプロパティを持つ。

名前	型	内容
CurrentAvailablePermits	long	空きトークンの数
CurrentQueuedCount	long	順番待ちの数
TotalFailedLeases	long	総Acquire失敗数
TotalSuccessfulLeases	long	総Acquire成功数

ReplenishingRateLimiter

後述するTokenBucket等、流量制御をトークンの補充(replenishment)で行うアルゴリズムを実装する時に使うベースクラス
派生クラスは以下を実装する必要がある。

• public abstract TimeSpan ReplenishmentPeriod { get; }
  • replenishment間隔
• public abstract bool IsAutoReplenishing { get; }
  • 自動でreplenishmentを行うかどうか
• public abstract bool TryReplenish()
  • replenishmentを手動実行する

標準で用意されているRateLimiter

ConcurrencyLimiter

セマフォのように、最初に最大で取得できる量を指定して、同時に取得できる数を制限するという、仕組み的には最も理解しやすいRateLimiter。
RateLimitLeaseのDisposeで許可できるpermitCountの回復を行うので、必ずDisposeを行う事。
使い方もほぼセマフォと一緒だが、取得する量をpermitCountで指定できるため、簡易なReaderWriterLockのように使うことも可能(WでpermitCount上限、Rでそれより少ない値を指定する)。
ただし、これを使って実装するRWロックはリエントラントではないので、デッドロックに注意すること。
ちゃんとしたものが使いたい場合は、 Microsoft.VisualStudio.Threading パッケージに色々とあるので、そちらを使うことを検討するといい。

設定値は、インスタンス作成時にConcurrencyRateLimiterOptionsで行う。
設定項目は以下

名前	型	意味
permitLimit	int	一度に受け入れられるpermitCountの数
queueProcessingOrder	QueueProcessingOrder	受け入れ待ちになったとき、後から入ったものが優先されるか、先に入ったものが優先されるか
queueLimit	int	受け入れ待ちをすることができるpermitCountの数。受け入れ待ちのキューの合計値がこれを超えると、キューから弾かれ、IsAcquiredがfalseにセットされて返ってくる

queueLimitから弾かれた時、先に入ったものか後から入ったもののどちらが弾かれるかは、queueProcessingOrderの設定値による。
Leaseが成功したときは特にメタデータは設定されないが、失敗したときは、TryGetMetadata(System.Threading.RateLimiting.MetadataName.ReasonPhrase.Name, out string reason)でメタデータを取得することができる。
これで、Acquireで失敗した時、permitLimitを超えたのか、それともqueueLimitも超えたのか、どちらなのかという情報を取得できる。

queueLimitを限りなく大きくすれば、深く考えずにAcquireAsyncすればその内順番が回ってくるという利点はあるが、
Limiterが保持している内部的なキューが限りなく大きくなる可能性があるので、その辺りは注意が必要だろう。

TokenBucketRateLimiter

トークンバケットアルゴリズムに基づいて制限を行う。

こちらは、同時にアクセスできる数を制限するのではなく、一定期間内にどれだけの量を許可するか(例: ネットワークデータ送受信量 bytes/sec)という使われ方をする。
一定の流量制限を実現しながら、スパイクの程度もある程度制御できるようなアルゴリズムとなっている。

設定値は以下のようになる。

名前	型	意味	対応するパラメーター
tokenLimit	int	replenishmentで蓄えられるtokenの最大数	b
queueProcessingOrder	QueueProcessingOrder	待ちになったとき、どちらが優先されるか	-
queueLimit	int	最大で待ち受けできる数	-
replenishmentPeriod	TimeSpan	どの程度の間隔で許可できるtokenが追加されるか	1/S
tokensPerPeriod	int	一度のreplenishで追加されるtokenの数	r/S
autoReplenishment	bool	自動で一定間隔ごとにtokenを追加するタスクを並行実行するか	-

queueLimitとqueueProcessingOrderの意味はConcurrencyLimiterと同じ。
この中で重要視されるのがtokenLimit,replenishmentPeriod,tokensPerPeriodとなるだろう。
rの値は、 tokensPerPeriod / replenishmentPeriod で決定される。
基本的にreplenishmentPeriodは、短くすれば詰まりが少なくなり、安定した処理が期待できるが、
あまり短い値にすると時刻の分解能に依存して間隔が安定しなくなり、またCPU負荷が高くなるので、
要件ごとに最適な値は変わってくると思う。

tokenLimitは、アイドル状態から一斉に要求があったとき、どの程度まで連続して許容するかという目安となる。
充分大きい値にしておけば、要求が不安定でも長期的に見て理想値に限りなく近付くという利点はあるが、
短期的に見ると処理がスパイクして安定しない場合があるという欠点がある。

autoReplenishment=falseを設定したとき、トークンは自動で補充されないので、自分でTokenBucketLimiter.TryReplenish()する必要がある。
この時、replenishmentPeriodとtokensPerPeriodの値が考慮された分だけトークンが補充される。
結局中で増分等は調整されるので、特に理由がない限りはデフォルトに任せてもいいと思う。

RateLimitLease.IsAcquire=falseで返ってきた時、RateLimitLease.TryGetMetadata<TimeSpan>(MetadataNames.RetryAfter, out TimeSpan retryAfter)で、何秒後にリトライすればいいかの目安が返ってくる。
必ずしもこの間隔分待たなければならないという事も無いが、この値に従えば理想値に近づくので、特に理由が無ければこの値に従って待つのが良い。
ただし、queueLimitを超えて弾き出された場合は、RetryAfterには値はセットされないので、TryGetMetadataの戻り値は必ずチェックするようにしよう。

FixedWindowRateLimiter

TokenBucketRateLimiterと似ているが、こちらは常に一定周期で限界値までトークンが補充されるものとなる。
仕組みとしてはTokenBucketよりも単純だが、その分スパイクの調整はTokenBucketより苦手で、Replenishmentの前後に要求が集中すると、
想定よりも大きい負荷になる場合がある。
仕組みとしては、 https://christina04.hatenablog.com/entry/rate-limiting-algorithm の"Fixed window counters"が詳しい。

オプション(FixedWindowRateLimiterOptions)で指定できるのは以下

名前	型	意味
permitLimit	int	一度に許可できる数
queueProcessingOrder	QueueProcessingOrder	キューで待つ時、新しいものを優先するか古いものを優先するか
queueLimit	int	キューで待てる最大の数
window	TimeSpan	補充が実行される周期
autoReplenishment	bool	裏で自動補充するタスクを動かすかどうか

TokenBucketRateLimiterOptionsからtokensPerPeriodを引いたようなイメージ。

SlidingWindowRateLimiter

https://christina04.hatenablog.com/entry/rate-limiting-algorithm の"Sliding window counters"を実現するもの。
FixedWindowでは、Replenishmentの前後で想定以上のスパイクが発生する可能性があったが、これはある程度その欠点を抑えることができるものとなっている。
本来のアルゴリズムでは、区間ごとの割合を計算しているが、浮動小数点の計算になるので、計算にある程度のコストがかかる。
実装では、割合計算を都度しているのではなく、一つの時間区間を更にいくつかのセグメントに分けて、それぞれのセグメントに許可されたリクエストカウントの合計を保持し、
一周回ったら回収するといった、ある程度計算コストを考えた実装となっている。
セグメントを増やせばより理想に近づいた値になるが、メモリが多く消費され、補充に関する負荷も増加するので、要調整。

負荷的にはこの記事で挙げた中で最も全体的な負荷が安定したアルゴリズムになるが、その代わり、一時的なアクセス集中を
あえて許容したい場合等に融通が効きにくいという特徴もある。
この辺りは、それぞれの場面に応じて最適なものを選ぶ必要がある。

文字だけで書くと推移が分かりにくいので例示すると、セグメント数3、100まで許可できるRateLimiterがあったとする。
[区間1の消費数],[区間2の消費数],[区間3の消費数]([使用可能な量]) という表記(太字は現在の区間)で書くと、

0,0,0(100)
↓50消費
50,0,0(50)
↓20消費
50,20,0(30)
↓10消費、50回復
0,20,10(70)
…
のような推移になる。

オプション( SlidingWindowRateLimiterOptions )で指定できるのは以下。

名前	型	意味
permitLimit	int	一度に許可できる数
queueProcessingOrder	QueueProcessingOrder	キューで待つ時、新しいものを優先するか古いものを優先するか
queueLimit	int	キューで待てる最大の数
window	TimeSpan	全体の周期
segmentsPerWindow	int	いくつの区間に分割するか
autoReplenishment	bool	裏で自動補充するタスクを動かすかどうか

補充は window / segmentsPerWindow の周期で行われる。

NoopLimiter

何もしないRateLimiter。要求があったとき、即時で許可のLeaseを返す。
ユニットテスト時のスタブやフォールバック等で使う。

PartitionedRateLimiter<TResource>

イメージ的には Dictionary<TKey, RateLimiter> だが、詳細を書くとまた別の記事が一つ作れそうなので、
記事を分けた

ASP.NET Coreでの活用

ASP.NET Coreでは、Microsoft.AspNetCore.RateLimitingパッケージを使うことにより、IIS等のHTTPサーバーに頼ることなく、流量制限を実現が可能。
RCまではnugetパッケージがあったが、正式リリースではフレームワーク(Microsoft.AspNetCore.App)の中に組み込まれているため、net7.0のaspnetcoreプロジェクトで何もしなくても使える。
サンプルによると、リクエストを一つの単位として制限をかける実装になっており、上記の各種RateLimiterを使うことが可能。

終わりに

RateLimiterは表立っては使われることは少ないかもしれないが、ネットワーク処理系のライブラリやフレームワーク内部では
多く使われることになるかもしれない(特にYARP)。

また、基本的な実装はインプロセスでオンメモリのものばかりだが、サードパーティ製の、複数プロセスorマシンが協調動作できるようなライブラリが
今後出てくるとまた色々と変わってくるかもしれない。
確か提案段階ではRedisによる実装等もあったかもしれない。

PartitionedRateLimiterについては、別途記事を作った
また、実際はMicrosoft.AspNetCore.RateLimiting経由で使うことが多いと予想されるため、この辺りも今後作っておきたい。