1. Homepage
  2. Flex Gateway ポリシー開発キット (PDK) (1.3)
  3. カスタムポリシーの開発
  4. Rust でのカスタムポリシー機能の実装

  5. 遅延関数、周期関数、同期関数の設定

遅延関数、周期関数、同期関数の設定

遅延、周期、および同期関数を有効にするために ​Clock​ を挿入してタイマーを作成します。各ポリシーでサポートされるタイマーは 1 つのみです。

各タイマーには、次の非同期メソッドがあります。

pub async fn next_tick(&self) -> bool;
pub async fn sleep(&self, interval: Duration) -> bool;

  • next_tick​: タイマーが識別できる最小時間スリープします。

  • sleep​: 目盛量スリープします。この場合、経過時間は指定した ​Duration​ 以上となります。たとえば、クロックの期間を 10 秒に設定した場合、​timer.sleep(Duration::from_secs(1))​ をコールすると 10 秒間スリープします。

どちらの関数も、指定した時間が経過すると ​true​ を返し、実行中にポリシーが適用解除または編集されると ​false​ を返します。

タイマーを設定する

タイマーをポリシーに挿入するには、ポリシー設定関数に ​Clock​ を挿入し、期間を指定してタイマーを定義します。

#[entrypoint]
async fn configure(
   launcher: Launcher,
   Configuration(bytes): Configuration,
   clock: Clock, // Inject the clock to be able to launch async tasks.
   client: HttpClient,
) -> Result<()> {


   // set the period between ticks of the timer.
   let timer = clock.period(Duration::from_secs(10));

タイマーを挿入した後、その参照を ​on_request​ 関数と ​on_response​ 関数と共有して遅延関数の作成を開始します。

    launcher
        .launch(on_request(|rs| request_filter(rs, &timer, &config)))
        .await?;

これで、次に示すように、​on_request​ 関数と ​on_response​ 関数でタイマー関数を待機できます。

    let slept = timer.sleep(duration).await;
    let tick = timer.next_tick().await;
遅延関数を起動するポリシープロジェクトの例については、 スパイクポリシーの例​を参照してください。

クロックをリリースする

異なるティック周期で新しいタイマーを作成するには、次のメソッドを使用して​クロック​をリリースします。

pub fn release(self) -> Clock;

次のコードスニペットでは、タイマーを作成してからクロックをリリースし、新しいタイマーを作成します。

    // Configure an initial tick period.
    let initial_timer = clock.period(Duration::from_millis(500));
    some_initial_task(&other_timer).await;
    // Configure another tick period.
    let other_timer = initial_timer.release().period(Duration::from_millis(3000);
    some_other_task(&other_timer).await;
    // Get the clock back.
    let clock = other_timer.release();

非同期タスクを起動する

タイマーを使用して、要求フローとは関係なくタスクを実行します。非同期タスクを実行する手順は、次のとおりです。

  1. async​ 関数内でタスクを定義します。

  2. ポリシー設定関数で、関数をコールします。ただし、返される将来は待機しません。

  3. ポリシー設定関数で、​on_request​ および ​on_response​ filter 関数を使用して ​launch​ 関数をコールします。ただし、返される将来は待機しません。

  4. join!​ マクロを使用して、タスク関数と ​launch​ 関数の両方の将来を同時に待機します。

    join​ マクロを使用するには、​cargo.toml​ ファイルに連動関係として 'futures' クレートを追加します。

各ワーカーは、タスクのコピーを実行します。関数内にループを設定して、周期タスクを実行します。たとえば、次のコード例は両方の関数を待機します。

let task = my_async_task(&timer, &config);
let launched = launcher.launch(filter);
let joined = join!(launched, task);

次に示すように、​my_async_task​ 関数にループを追加すると周期タスクが作成されます。

async fn my_async_task(timer: &Timer, config: &Config) {
    // While the policy is still running.
    // Wait for the next cycle.
    while timer.next_tick().await {
        // Execute periodic task
    }
}

複数の非同期タスクを起動するには、次に示すようにタイマーの参照を ​async​ 関数に送信し、それらを ​launch​ タスクと結合します。

let task1 = my_async_task1(&timer, &config);
let task2 = my_async_task2(&timer, &config);
let launched = launcher.launch(filter);
let joined = join!(launched, task1, task2);
非同期の周期タスクを起動するポリシープロジェクトの例については、 メトリクスポリシーの例​ を参照してください。

ワーカー間で非同期タスクを同期する

PDK ​LockBuilder​ を使用してワーカーを同期します。ロックを使用すると、1 つの期間にタスクを完了するワーカーは 1 つのみになります。各ポリシーは、複数のロックをサポートします。

ロックは、ポリシーが使用制限のある HTTP サービスをコンシュームする場合に便利です。1 つのワーカーがデータをフェッチし、そのデータを他のワーカーと共有して、サービス要求を減らすことができます。

ロックを設定する手順は、次のとおりです。

  1. LockBuilder​ をポリシー設定関数に挿入します。

    #[entrypoint]
async fn configure(
    launcher: Launcher,
    Configuration(bytes): Configuration,
    clock: Clock,        // Inject the clock to be able to launch async tasks.
    lock: LockBuilder,   // Inject the lock to be able to synchronize the workers.
) -> Result<()> {

  2. ロックを設定し、その参照を非同期関数と共有します。

        let lock = lock
        .new(ID.to_string())
        .expiration(Duration::from_secs(20))
        .build();
    let task = my_async_task(&timer, &lock);

    非同期タスクがタスク内で追加の ​async​ 関数コールを実行する場合は、ロックの ​expiration​ を、ネストされた ​async​ 関数コールを完了するために必要な合計時間よりも長く設定します。

  3. 非同期関数内で、目的のタスクを実行する前にロックを取得します。他のワーカーにロックが設定されていない場合、現在のワーカーがロックを取得します。それ以外の場合は、次に示すようにタスクの実行を省略し、次のタスクループを待機する必要があります。

    async fn my_async_task(
    timer: &Timer,
    lock: &TryLock,
) {
   while timer.next_tick().await {
     if let Some(acquired) = lock.try_lock() {
       // execute task.
     }
   }
}

    ロックを取得した後、結果を待機する必要があるコードを実行する場合は、​refresh_lock​ 関数をコールして、ロックの有効期限が切れていないことを確認する必要があります。ロックの有効期限が切れている場合、ワーカーは無応答と見なされます。別のワーカーが同じタスク実行を実行しているため、タスクの実行を中止します。

非同期の周期タスクを同期し、ワーカー間で情報を共有するポリシープロジェクトの例については、 ブロックポリシーの例​を参照してください。

関連情報