Linux カーネルおさんぽマップ〜時計編〜

本稿では，Linux カーネルで扱える時計のうち，PTP（Precision Time Protocol）という IEEE 1588 で策定されている高精度な時計をクロックソース[1]にする場合について追っていこうと思います。

Linux における時刻同期：NTP と PTP

Linux や Windows，macOS といった現在の OS のシステムクロックは，一般的にはインターネットを通じて NTP（Network Time Protocol）が同期するようになっています。
NTP を使っていればインターネットに接続されている際には大きく時刻がズレることはなく，たまにネットワーク接続が失われていてもクォーツ時計は安物の振り子式ほど時間はズレません。

では，NTP は設計としてどのくらいの誤差が考えられているのでしょうか。NTP の設計者 David L. Mills 博士は自身が実装した PDP-11 向けの OS「Fuzzball」上の実験でそれを示しており，答えは数十ミリ秒です[2]。現代のマシンと OS ならば数ミリ秒までは迫れるかもしれません。ちなみにこの論文が発表された翌年には NTP version 2 が RFC から発行されています[3]。（現在は version 4。）

NTP は日常的な利用には十分な精度があります。しかし，マイクロ秒やナノ秒といった単位での高精度さが目的ではありません。日本の NTP のサーバーとして有名な NICT（情報通信研究機構）は日本標準時の決定をしており原子時計も所有していますが，インターネットを介して NICT などのサーバーにアクセスしている時点でミリ秒やひどいときは数秒の遅れが発生するでしょう。LAN に NTP サーバーがあったとしても，アルゴリズムとして数ミリ秒の遅延を許容しています。そこで高精度な時刻同期のために存在するのが PTP です。

PTP とは

PTP とは，ネットワークデバイス内の時計を高精度に同期する仕組みです。グランドマスタークロックと呼ばれる時計をマスターとして設置し，LAN の NIC たちがそれを元に同期するといったような仕組みになります。これによって，パケットのタイムスタンプをハードウェアで高精度に打つことが可能になったり，この PTP をクロックソースとしてシステムクロックを同期したりできます。

ixgbe と PTP

Intel の 10GbE のネットワークインターフェイスとして，ドライバ名から ixgbe と呼ばれるものがあります。Intel 82599 や Intel X540 などが該当します。今回は Intel X550[4 sec. 7.7]について追っていこうと思います。

前準備・Intel アーキテクチャと割り込みについて

ソースコードを追う前にまず，PCIe 機器はどのように CPU に接続されどのように扱われるかを知らなければなりません。

図１．Intel アーキテクチャと割り込みの関係についてのブロックダイアグラム図

Intel CPU はおおむね上図のような形になっています[5 sec. 10.1]。大昔は Intel 8259 という割り込みコントローラーが直接 CPU に接続されて，周辺機器の割り込み信号は全部このコントローラーに接続されていましたが，現在では CPU の各コアに Local APIC という割り込みコントローラーがあり，チップセットのほうに割り込み I/O APIC という割り込みコントローラーがあります。このように分離されているおかげで，周辺機器からの割り込み信号は I/O APIC が受け取ってからどのコアに割り込みを配送するか自由に設定できますし，コア同士の通信は Local APIC を通じて IPI を送ることで可能となっています。

PCI デバイスは I/O APIC へ割り込み信号を送ることも可能ですが，もうひとつ，MSI という割り込みも可能です[6 sec. 6.8]。これは，割り込みコントローラーのピンに割り込み線を接続して INTx で割り込む（pin-based 割り込み）のではなく，デバイスにメモリに直接割り込みメッセージを書くことによって CPU コアの Local APIC へ割り込みが送られます。

MSI での割り込みは pin-based の割り込みと違い，ピンが少なくてすんだりパフォーマンス上有利だったりとメリットが多く，また，PCIe デバイスでは MSI やその拡張である MSI-X が必須になっています。

ソースコードを読もう

今回は v4.17-rc7 のソースコードを読みます。ixgbe のソースコードは，drivers/net/ethernet/intel/ixgbe/にあります。以下断わりなくソースコードのファイル名を出した場合はこのディレクトリを起点に考えてください。

Linux のクロックソースとして ixgbe の PTP を設定するとなると，ixgbe のエントリポイントで割り込みベクタを設定する部分を読む必要がありそうです。そこで，irq や msix といった割り込みに関連しそうな処理をエントリポイントから探してみます。

まずはエントリポイントそのものを探さなければなりません。ixgbe_main.cといういかにもなソースコードから entry point という語で探すと，6584 行目，ixgbe_open()が見つかります。

int ixgbe_open(struct net_device *netdev)
{
     struct ixgbe_adapter *adapter = netdev_priv(netdev);
     struct ixgbe_hw *hw = &adapter->hw;
      int err, queues;

     /* disallow open during test */
     if (test_bit(__IXGBE_TESTING, &adapter->state))
         return -EBUSY;

     netif_carrier_off(netdev);

    /* allocate transmit descriptors */
     err = ixgbe_setup_all_tx_resources(adapter);
     if (err)
         goto err_setup_tx;

     /* allocate receive descriptors */
     err = ixgbe_setup_all_rx_resources(adapter);
     if (err)
         goto err_setup_rx;

     ixgbe_configure(adapter);

     err = ixgbe_request_irq(adapter);

このコードの 6088 行目でixgbe_request_irq(adapter);という呼び出しがあります。この関数は，同じソースコードに短い定義があります。3418 行目です。

static int ixgbe_request_irq(struct ixgbe_adapter *adapter)
 {
     struct net_device *netdev = adapter->netdev;
     int err;

     if (adapter->flags & IXGBE_FLAG_MSIX_ENABLED)
         err = ixgbe_request_msix_irqs(adapter);
     else if (adapter->flags & IXGBE_FLAG_MSI_ENABLED)
         err = request_irq(adapter->pdev->irq, ixgbe_intr, 0,
                   netdev->name, adapter);
     else
         err = request_irq(adapter->pdev->irq, ixgbe_intr, IRQF_SHARED,
                   netdev->name, adapter);

     if (err)
         e_err(probe, "request_irq failed, Error %d\n", err);

     return err;
 }

3424 行目に ixgbe_request_msix_irqs(adapter) がありますね。ここで MSI-X の割り込みを登録しているのではないか，と推測されますね。では，実際に確かめます。同一ファイル 3274 行目。

static int ixgbe_request_msix_irqs(struct ixgbe_adapter *adapter)
  {
      struct net_device *netdev = adapter->netdev;
      unsigned int ri = 0, ti = 0;
      int vector, err;

      for (vector = 0; vector < adapter->num_q_vectors; vector++) {
          struct ixgbe_q_vector *q_vector = adapter->q_vector[vector];
          struct msix_entry *entry = &adapter->msix_entries[vector];

          if (q_vector->tx.ring && q_vector->rx.ring) {
              snprintf(q_vector->name, sizeof(q_vector->name),
                   "%s-TxRx-%u", netdev->name, ri++);
              ti++;
          } else if (q_vector->rx.ring) {
              snprintf(q_vector->name, sizeof(q_vector->name),
                   "%s-rx-%u", netdev->name, ri++);
          } else if (q_vector->tx.ring) {
              snprintf(q_vector->name, sizeof(q_vector->name),
                   "%s-tx-%u", netdev->name, ti++);
          } else {
              /* skip this unused q_vector */
              continue;
          }
          err = request_irq(entry->vector, &ixgbe_msix_clean_rings, 0,
                    q_vector->name, q_vector);
          if (err) {
              e_err(probe, "request_irq failed for MSIX interrupt "
                    "Error: %d\n", err);
              goto free_queue_irqs;
          }
          /* If Flow Director is enabled, set interrupt affinity */
          if (adapter->flags & IXGBE_FLAG_FDIR_HASH_CAPABLE) {
              /* assign the mask for this irq */
              irq_set_affinity_hint(entry->vector,
                            &q_vector->affinity_mask);
          }
      }

      err = request_irq(adapter->msix_entries[vector].vector,
                ixgbe_msix_other, 0, netdev->name, adapter);

割り込みベクタを登録していると思しき関数呼び出し，request_irq() が二箇所ありますね？これはそれぞれどういうことでしょう。ひとつめの request_irq() （3298 行目）は 3280 行目からのイテレーターの中にあります。このイテレーターではデバイスの構造体である adapter にある q_vector というものをイテレートしているようです。

/* MAX_Q_VECTORS of these are allocated,
 * but we only use one per queue-specific vector.
 */
struct ixgbe_q_vector {
    struct ixgbe_adapter *adapter;
#ifdef CONFIG_IXGBE_DCA
    int cpu;        /* CPU for DCA */
#endif
    u16 v_idx;      /* index of q_vector within array, also used for
                 * finding the bit in EICR and friends that
                 * represents the vector for this ring */
    u16 itr;        /* Interrupt throttle rate written to EITR */
    struct ixgbe_ring_container rx, tx;

    struct napi_struct napi;
    cpumask_t affinity_mask;
    int numa_node;
    struct rcu_head rcu;    /* to avoid race with update stats on free */
    char name[IFNAMSIZ + 9];

    /* for dynamic allocation of rings associated with this q_vector */
    struct ixgbe_ring ring[0] ____cacheline_internodealigned_in_smp;
};

adapter->q_vector のデータ構造の定義を ixgbe.h から探してみるとこのようになっていることから，送受信のためのキューとその設定の構造体でしょう。ひとつの adapter に複数 q_vector が登録できるのはマルチキューのためですかね。さて，ixgbe_request_msix_irqs()に戻ってみてみると，3284 行目から 3294 行目までの条件分岐で vector が rx.ring か tx.ring かを見ているため，これが，送受信のためのリングバッファが作られているのならそれに関連した割り込みベクタを登録しているのだということが推測できます。時計の処理には関係なさそうなので，もうひとつの request_irq() をみてみましょう。

3313 行目のもうひとつの request_irq() では ixgbe_msix_other という関数の名前を持ってきています。この関数はやはり同じソースコード，ixgbe_main.c の 3109 行目に存在しています。

static irqreturn_t ixgbe_msix_other(int irq, void *data)
  {
      struct ixgbe_adapter *adapter = data;
      struct ixgbe_hw *hw = &adapter->hw;
      u32 eicr;

      /*
       * Workaround for Silicon errata.  Use clear-by-write instead
       * of clear-by-read.  Reading with EICS will return the
       * interrupt causes without clearing, which later be done
       * with the write to EICR.
       */
      eicr = IXGBE_READ_REG(hw, IXGBE_EICS);

      /* The lower 16bits of the EICR register are for the queue interrupts
       * which should be masked here in order to not accidentally clear them if
       * the bits are high when ixgbe_msix_other is called. There is a race
       * condition otherwise which results in possible performance loss
       * especially if the ixgbe_msix_other interrupt is triggering
       * consistently (as it would when PPS is turned on for the X540 device)
       */
      eicr &= 0xFFFF0000;

      IXGBE_WRITE_REG(hw, IXGBE_EICR, eicr);

...

      （略）

...


      if (unlikely(eicr & IXGBE_EICR_TIMESYNC))
          ixgbe_ptp_check_pps_event(adapter);

      /* re-enable the original interrupt state, no lsc, no queues */
      if (!test_bit(__IXGBE_DOWN, &adapter->state))
          ixgbe_irq_enable(adapter, false, false);

      return IRQ_HANDLED;
  }

やりました，ついに見つけました。3185，3186 行目，EICR というレジスタを確認して，ixgbe_ptp_check_pps_event(adapter) を呼び出しています。IXGBE_EICR_TIMESYNC というのは，レジスタ EIRC と論理積でテストをしているためレジスタの何らかのフラグではないかと考えられます。 X550 のデータシートの EICR レジスタのところ[4 sec. 8.2.2.6.1]を確認すると，割り込み原因が書き込まれるレジスタであることがわかります。どうやら 24-bit 目が TIMESYNC レジスタのようで，NIC に時刻同期を要求する割り込みが来るとこのフラグが立つようです。

さて，この ixgbe_ptp_check_pps_event() を探してみると，ixgbe_main.c にはありません。ついにこのファイルから離れ他のファイルを探してみると，ixgbe_ptp.cの 559 行目に見つかります。

/**
 * ixgbe_ptp_check_pps_event
 * @adapter: the private adapter structure
 *
 * This function is called by the interrupt routine when checking for
 * interrupts. It will check and handle a pps event.
 */
void ixgbe_ptp_check_pps_event(struct ixgbe_adapter *adapter)
{
    struct ixgbe_hw *hw = &adapter->hw;
    struct ptp_clock_event event;

    event.type = PTP_CLOCK_PPS;

    /* this check is necessary in case the interrupt was enabled via some
     * alternative means (ex. debug_fs). Better to check here than
     * everywhere that calls this function.
     */
    if (!adapter->ptp_clock)
        return;

    switch (hw->mac.type) {
    case ixgbe_mac_X540:
        ptp_clock_event(adapter->ptp_clock, &event);
        break;
    default:
        break;
    }
}

NIC のドライバなのでちょっと混乱しそうになるのですが，ここで言う pps とは packet per second ではなく，pulse per second signal です。1 秒あたりのパルス数を表示する単位，pulses per second とも違います。キッカリ 1 秒ごとにパルスを発するような信号のことを言い，この pulse per second signal のための Unix-like OS の API が RFC に策定されていたりします[7]。

この関数，ptp_clock_event() を呼び出していますが，X540 の場合のみ対応しているようです。X550 は対応していないのかしら。ここで終わってしまうのもなんなので，この先の処理も追ってみます。

ここからは ixgbe のコードを一気に離れ，ptp のコードに入ります。drivers/ptp/ptp_clock.c の 325 行目を見てみましょう。

void ptp_clock_event(struct ptp_clock *ptp, struct ptp_clock_event *event)
  {
      struct pps_event_time evt;

      switch (event->type) {

      case PTP_CLOCK_ALARM:
          break;

      case PTP_CLOCK_EXTTS:
          enqueue_external_timestamp(&ptp->tsevq, event);
          wake_up_interruptible(&ptp->tsev_wq);
          break;

      case PTP_CLOCK_PPS:
          pps_get_ts(&evt);
          pps_event(ptp->pps_source, &evt, PTP_PPS_EVENT, NULL);
          break;

      case PTP_CLOCK_PPSUSR:
          pps_event(ptp->pps_source, &event->pps_times,
                PTP_PPS_EVENT, NULL);
          break;
      }
  }
  EXPORT_SYMBOL(ptp_clock_event);

ptp_clock_event() は PTP_CLOCK_PPS というイベントで呼び出されていました。そこでみてみると，pps_event() という API が呼び出されています。この関数の説明は Documentations/pps/pps.txt にも載ってまして，割り込みハンドラのような pps 信号へのイベントはこれを使い登録するそうです。この関数の定義は drivers/pps/kapi.c の 172 行目にあり，

/* pps_event - register a PPS event into the system
 * @pps: the PPS device
 * @ts: the event timestamp
 * @event: the event type
 * @data: userdef pointer
 *
 * This function is used by each PPS client in order to register a new
 * PPS event into the system (it's usually called inside an IRQ handler).
 *
 * If an echo function is associated with the PPS device it will be called
 * as:
 *  pps->info.echo(pps, event, data);
 */
void pps_event(struct pps_device *pps, struct pps_event_time *ts, int event,
          void *data)
{
    unsigned long flags;
    int captured = 0;
    struct pps_ktime ts_real = { .sec = 0, .nsec = 0, .flags = 0 };

...

      （略）

...


    pps_kc_event(pps, ts, event);

    /* Wake up if captured something */
    if (captured) {
        pps->last_ev++;
        wake_up_interruptible_all(&pps->queue);

        kill_fasync(&pps->async_queue, SIGIO, POLL_IN);
    }

    spin_unlock_irqrestore(&pps->lock, flags);
}
EXPORT_SYMBOL(pps_event);

さらに pps_kc_event() が呼ばれています。この関数の定義は drivers/pps/kc.c の 112 行目であり，pps 信号のイベントに対してただ hardpps() を呼び出すだけのようです。

/* pps_kc_event - call hardpps() on PPS event
 * @pps: the PPS source
 * @ts: PPS event timestamp
 * @event: PPS event edge
 *
 * This function calls hardpps() when an event from bound PPS source occurs.
 */
void pps_kc_event(struct pps_device *pps, struct pps_event_time *ts,
         int event)
{
     unsigned long flags;

     /* Pass some events to kernel consumer if activated */
     spin_lock_irqsave(&pps_kc_hardpps_lock, flags);
     if (pps == pps_kc_hardpps_dev && event & pps_kc_hardpps_mode)
          hardpps(&ts->ts_real, &ts->ts_raw);
     spin_unlock_irqrestore(&pps_kc_hardpps_lock, flags);
}

そろそろ脳内の呼び出しスタックも深くなりすぎていい加減うんざりしてきたところだと思います。hardpps() のコードともなるともはやドライバーのソースですらありません。kernel/time/timekeeping.c 2347 行目，

#ifdef CONFIG_NTP_PPS
/**
 * hardpps() - Accessor function to NTP __hardpps function
 */
void hardpps(const struct timespec64 *phase_ts, const struct timespec64 *raw_ts)
{
    unsigned long flags;

    raw_spin_lock_irqsave(&timekeeper_lock, flags);
    write_seqcount_begin(&tk_core.seq);

        __hardpps(phase_ts, raw_ts);

    write_seqcount_end(&tk_core.seq);
    raw_spin_unlock_irqrestore(&timekeeper_lock, flags);
}
EXPORT_SYMBOL(hardpps);
#endif /* CONFIG_NTP_PPS */

これもまたロックなどをしていますが関数を呼び出しているのが主な処理であり，__hardpps()という NTP の処理のための内部関数を呼んでいます。 kernel/time/ntp.c の 973 行目，

/*
 * __hardpps() - discipline CPU clock oscillator to external PPS signal
 *
 * This routine is called at each PPS signal arrival in order to
 * discipline the CPU clock oscillator to the PPS signal. It takes two
 * parameters: REALTIME and MONOTONIC_RAW clock timestamps. The former
 * is used to correct clock phase error and the latter is used to
 * correct the frequency.
 *
 * This code is based on David Mills's reference nanokernel
 * implementation. It was mostly rewritten but keeps the same idea.
 */
void __hardpps(const struct timespec64 *phase_ts, const struct timespec64 *raw_ts)
{
    struct pps_normtime pts_norm, freq_norm;

    pts_norm = pps_normalize_ts(*phase_ts);
        
        ...

コメントによるとどうやら，hardpps() というのはまたもや出てきた NTP の著者の David L. Mills 博士の成果である nanokernel の実装が初出のものだそうです。 nanokernel というのを調べてみると，アメリカの航海学会で開催されている Precise Time and Time Interval (PTTI) Meeting（正確な時間と時間間隔の会合）のうち第 32 回目のときに報告された論文が元のようですね[8]。なんか意外な学会ですが，PTTI 2000 の URL は U.S. Navy のドメインでホストされていたのでなんか納得。海軍の作戦行動とかシステムはやっぱり時刻同期も重要なのでしょう。

この論文を読んでも良いのですが，どうやらhardpps()のアルゴリズムについては NTP.org にある解説のほうが要約されていてわかりやすそうです[9]。システムの時計での pps と外部信号（今回は PTP による pps ですね）を使って，システムの時計のオフセットのズレや周期のズレといったものを最小化しつつ pps に同期するようです。

次回予告

お気づきかもしれませんが，途中で割り込みハンドラ登録関数であるところの request_irq() の処理についてすっ飛ばしました。なので，たとえば NIC に PTP から時刻同期の信号が入り OS に割り込みが MSI-X を通じて飛んだとして，先程まで読んできた hardpps() を呼びだすような割り込みベクタ ixgbe_msix_other がどのように呼び出されるかはまだわかっていません。この割り込みの呼び出しにはどのような処理がありどのくらいオーバーヘッドが見込まれるでしょうか。次回をお楽しみください。

参考文献

[1] IEEE，“IEEE 1588^TM Standard for A Precision Clock Synchronization Protocol for Networked Measurement and Control Systems,” URL: https://www.nist.gov/el/intelligent-systems-division-73500/ieee-1588.
[2] D. L. Mills, “The Fuzzball,” in Proc. of Symposium Proceedings on Communications Architectures and Protocols, Aug. 1988, pp. 115-122. URL: https://www.eecis.udel.edu/~mills/database/papers/fuzz.pdf.
[3] D. L. Mills, “Internet Time Synchronization: the Network Time Protocol,” IETF, Oct. 1989, URL: https://tools.ietf.org/pdf/rfc1129.pdf.
[4] Intel, “Intel®︎ Ethernet Controller X550 Datasheet,” URL: https://www.intel.com/content/dam/www/public/us/en/documents/datasheets/ethernet-x550-datasheet.pdf.
[5] Intel, “Intel® 64 and IA-32 architectures software developer's manual volume 3A: System programming guide, part 1,” URL: https://software.intel.com/en-us/articles/intel-sdm.
[6] PCI-SIG, “PCI Local Bus Specification Revision 3.0,” URL: https://www.xilinx.com/Attachment/PCI_SPEV_V3_0.pdf.
[7] J. Mogul, D. Mills, J. Brittenson, J. Stone and U. Windl “Pulse-Per-Second API for UNIX-like Operating Systems, Version 1.0,” IETF, Mar. 2000, URL: https://tools.ietf.org/html/rfc2783.
[8] D. L. Mills and P.-H. Kamp, “THE NANOKERNEL,” in Proc. of 32^nd Annual Precise Time and Time Interval Meeting, Nov. 2000, pp. 423-430. URL: https://www.eecis.udel.edu/~mills/database/papers/nanokernel.pdf.
[9] NTP.org, “5.2 The Kernel Dicipline,” URL: http://www.ntp.org/ntpfaq/NTP-s-algo-kernel.htm#Q-ALGO-KERNEL-HARDPPS.

技術者見習いの独り言

このブログを検索