RCU hlist_nulls로 목록과 객체 보호하기

SLAB_TYPESAFE_BY_RCU 목록의 탐색 문제

이 문서는 `SLAB_TYPESAFE_BY_RCU`로 할당된 객체와 읽기 위주 연결 목록을 `hlist_nulls`로 보호하는 방법을 설명한다. 기본 목록 RCU 사용법은 `listRCU.rst`를 먼저 참고한다. 예제 객체에는 `hlist_node`, 원자적 참조 카운터 `refcnt`, 검색 키 `key`가 있다.

일반 hlist를 잠금 없이 찾을 때는 `rcu_read_lock()` 안에서 키를 검색하고 `try_get_ref()`로 살아 있는 객체의 참조를 얻어야 한다. `SLAB_TYPESAFE_BY_RCU`는 slab 페이지의 수명만 늦출 뿐 객체 자체의 재사용은 막지 않으므로, 참조 획득 뒤 `obj->key`를 다시 검사해야 한다. 참조 획득에 실패하거나 키가 달라졌다면 참조와 읽기 잠금을 놓고 처음부터 다시 찾는다.

전통적인 `hlist_for_each_entry_rcu()`만으로는 충분하지 않다. 작성자가 해시 계산과 `next` 접근 사이에 객체를 다른 목록의 끝으로 옮기면, 독자는 새 목록의 NULL 끝을 보고 원래 체인의 나머지를 놓칠 수 있다. 따라서 `lockless_lookup()`은 현재 노드의 키를 검사하기 전에 `next`를 미리 읽고 `smp_rmb()`를 수행한 뒤 그 저장값으로 다음 노드에 진행해야 한다.

Corey Minyard의 설명처럼 핵심은 객체가 다른 목록으로 이동했다는 사실을 일반 NULL 끝 표시만으로는 구별할 수 없다는 점이다. 올바른 장벽과 함께 `next`를 선취하면 키 확인 전에 원래 진행 방향을 확보할 수 있다.

.. SPDX-License-Identifier: GPL-2.0

=================================================
Using RCU hlist_nulls to protect list and objects
=================================================

This section describes how to use hlist_nulls to
protect read-mostly linked lists and
objects using SLAB_TYPESAFE_BY_RCU allocations.

Please read the basics in listRCU.rst.

Using 'nulls'
=============

Using special makers (called 'nulls') is a convenient way
to solve following problem.

Without 'nulls', a typical RCU linked list managing objects which are
allocated with SLAB_TYPESAFE_BY_RCU kmem_cache can use the following
algorithms.  Following examples assume 'obj' is a pointer to such
objects, which is having below type.

::

  struct object {
    struct hlist_node obj_node;
    atomic_t refcnt;
    unsigned int key;
  };

1) Lookup algorithm
-------------------

::

  begin:
  rcu_read_lock();
  obj = lockless_lookup(key);
  if (obj) {
    if (!try_get_ref(obj)) { // might fail for free objects
      rcu_read_unlock();
      goto begin;
    }
    /*
    * Because a writer could delete object, and a writer could
    * reuse these object before the RCU grace period, we
    * must check key after getting the reference on object
    */
    if (obj->key != key) { // not the object we expected
      put_ref(obj);
      rcu_read_unlock();
      goto begin;
    }
  }
  rcu_read_unlock();

Beware that lockless_lookup(key) cannot use traditional hlist_for_each_entry_rcu()
but a version with an additional memory barrier (smp_rmb())

::

  lockless_lookup(key)
  {
    struct hlist_node *node, *next;
    for (pos = rcu_dereference((head)->first);
         pos && ({ next = pos->next; smp_rmb(); prefetch(next); 1; }) &&
         ({ obj = hlist_entry(pos, typeof(*obj), obj_node); 1; });
         pos = rcu_dereference(next))
      if (obj->key == key)
        return obj;
    return NULL;
  }

And note the traditional hlist_for_each_entry_rcu() misses this smp_rmb()::

  struct hlist_node *node;
  for (pos = rcu_dereference((head)->first);
       pos && ({ prefetch(pos->next); 1; }) &&
       ({ obj = hlist_entry(pos, typeof(*obj), obj_node); 1; });
       pos = rcu_dereference(pos->next))
    if (obj->key == key)
      return obj;
  return NULL;

Quoting Corey Minyard::

  "If the object is moved from one list to another list in-between the
  time the hash is calculated and the next field is accessed, and the
  object has moved to the end of a new list, the traversal will not
  complete properly on the list it should have, since the object will
  be on the end of the new list and there's not a way to tell it's on a
  new list and restart the list traversal. I think that this can be
  solved by pre-fetching the "next" field (with proper barriers) before
  checking the key."

삽입과 제거의 순서 보장

삽입자는 독자가 새 `obj_node.next`와 이전 `obj->key`를 함께 관찰하지 못하도록 해야 한다. 객체가 한 체인에서 삭제되어 비어 있던 다른 체인으로 옮겨지면 새 `next`는 NULL일 수 있고, 순서가 어긋난 독자는 원래 체인의 후속 항목을 놓쳤다는 사실을 감지하지 못한다.

새 항목은 중간이나 끝이 아니라 목록 머리에 넣는다. `kmem_cache_alloc()` 뒤 체인 잠금을 잡고 키를 기록하며, `atomic_set_release(&obj->refcnt, 1)`로 키 초기화가 참조 카운터 공개보다 먼저 보이게 한다. 이어 `hlist_add_head_rcu()`로 게시하고 잠금을 푼다.

삭제는 표준 RCU hlist 삭제를 사용할 수 있다. 마지막 참조를 놓는 쪽이 체인 잠금 아래 `hlist_del_init_rcu()`를 수행한 뒤 `kmem_cache_free()`한다. 단, `SLAB_TYPESAFE_BY_RCU`에서는 삭제된 객체가 grace period가 끝나기 전에도 매우 빠르게 다른 객체로 재사용될 수 있다는 점을 항상 전제로 해야 한다.

2) Insertion algorithm
----------------------

We need to make sure a reader cannot read the new 'obj->obj_node.next' value
and previous value of 'obj->key'. Otherwise, an item could be deleted
from a chain, and inserted into another chain. If new chain was empty
before the move, 'next' pointer is NULL, and lockless reader can not
detect the fact that it missed following items in original chain.

::

  /*
   * Please note that new inserts are done at the head of list,
   * not in the middle or end.
   */
  obj = kmem_cache_alloc(...);
  lock_chain(); // typically a spin_lock()
  obj->key = key;
  atomic_set_release(&obj->refcnt, 1); // key before refcnt
  hlist_add_head_rcu(&obj->obj_node, list);
  unlock_chain(); // typically a spin_unlock()


3) Removal algorithm
--------------------

Nothing special here, we can use a standard RCU hlist deletion.
But thanks to SLAB_TYPESAFE_BY_RCU, beware a deleted object can be reused
very very fast (before the end of RCU grace period)

::

  if (put_last_reference_on(obj) {
    lock_chain(); // typically a spin_lock()
    hlist_del_init_rcu(&obj->obj_node);
    unlock_chain(); // typically a spin_unlock()
    kmem_cache_free(cachep, obj);
  }

nulls 끝 표시로 추가 장벽 피하기

`hlist_nulls`를 사용하면 잠금 없는 검색의 추가 `smp_rmb()`를 피할 수 있다. 해시 테이블 각 슬롯의 번호를 해당 체인의 특수 끝 표시인 nulls 값으로 저장하면, 검색이 체인 끝에 도달했을 때 관찰한 nulls 값과 시작 슬롯을 비교해 삭제나 이동 경합을 검출할 수 있다.

끝의 nulls 값이 예상한 슬롯 번호와 다르면 독자가 다른 체인으로 이동한 항목을 만난 것이므로 검색을 처음부터 다시 시작한다. 객체가 같은 체인 안에서 이동했다면 목록을 한 번 더 훑을 수는 있어도 정확성에는 문제가 없다. 필드 형식도 `struct hlist_node`에서 `struct hlist_nulls_node`로 바뀐다.

검색은 `hlist_nulls_for_each_entry_rcu()`로 항목을 순회하고, 키가 맞으면 참조 획득과 키 재검사를 수행한다. 끝에 도달했을 때 `get_nulls_value(node) != slot`이면 잠금과 참조를 정리한 뒤 재시도하며, 값이 맞으면 찾지 못한 결과를 확정한다.

삽입 절차와 release 순서는 일반 hlist 예제와 같지만 `hlist_nulls_add_head_rcu()`를 사용한다. 이 API 역시 새 항목을 목록 머리에 게시한다.

--------------------------------------------------------------------------

Avoiding extra smp_rmb()
========================

With hlist_nulls we can avoid extra smp_rmb() in lockless_lookup().

For example, if we choose to store the slot number as the 'nulls'
end-of-list marker for each slot of the hash table, we can detect
a race (some writer did a delete and/or a move of an object
to another chain) checking the final 'nulls' value if
the lookup met the end of chain. If final 'nulls' value
is not the slot number, then we must restart the lookup at
the beginning. If the object was moved to the same chain,
then the reader doesn't care: It might occasionally
scan the list again without harm.

Note that using hlist_nulls means the type of 'obj_node' field of
'struct object' becomes 'struct hlist_nulls_node'.


1) lookup algorithm
-------------------

::

  head = &table[slot];
  begin:
  rcu_read_lock();
  hlist_nulls_for_each_entry_rcu(obj, node, head, obj_node) {
    if (obj->key == key) {
      if (!try_get_ref(obj)) { // might fail for free objects
        rcu_read_unlock();
        goto begin;
      }
      if (obj->key != key) { // not the object we expected
        put_ref(obj);
        rcu_read_unlock();
        goto begin;
      }
      goto out;
    }
  }

  // If the nulls value we got at the end of this lookup is
  // not the expected one, we must restart lookup.
  // We probably met an item that was moved to another chain.
  if (get_nulls_value(node) != slot) {
    put_ref(obj);
    rcu_read_unlock();
    goto begin;
  }
  obj = NULL;

  out:
  rcu_read_unlock();

2) Insert algorithm
-------------------

Same to the above one, but uses hlist_nulls_add_head_rcu() instead of
hlist_add_head_rcu().

::

  /*
   * Please note that new inserts are done at the head of list,
   * not in the middle or end.
   */
  obj = kmem_cache_alloc(cachep);
  lock_chain(); // typically a spin_lock()
  obj->key = key;
  atomic_set_release(&obj->refcnt, 1); // key before refcnt
  /*
   * insert obj in RCU way (readers might be traversing chain)
   */
  hlist_nulls_add_head_rcu(&obj->obj_node, list);
  unlock_chain(); // typically a spin_unlock()