FreeBSD 基礎建設 - Synchronization

by thinker

關鍵字:

Synchroniztion ；同步機制， multi-task kernel 的基礎建設之一。以下介紹 $FreeBSD$ 如何建構其同步機制 mutex 。

=== Atomic OP ===
Atomic operator 泛指各種不會在執行中被中斷的指令，這裡指各種用於實作 synchronization 所需的 atomic 指令，例如 read and set 或 read and clear 指令。這類指令是實作各種 synchronization 機制的基礎，$FreeBSD$ 定義一組基本 atomic 指令，在不同平臺上各自實作，以實現跨平臺的 synchronization 機制。

這些 atomic OPs 都定義在 machine/atomic.h ， machine 實際上對像到 <mach>/include/ ，如 i386/include/ 、 arm/include/ 。 Atomic 指令包函下面所列的存取指令:
* atomic_set_*(P,V)
* atomic_clear_*(P,V)
* atomic_add_*(P,V)
* atomic_substract_*(P,V)
* atomic_cmpset_*(P,E,V)
* atomic_fetchadd_*(P,V)
* atomic_load_acq_*(P)
* atomic_store_rel_*(P,V)
* atomic_readandclear_*(P,V)
* atomic_set_acq_*(P,V)
* atomic_set_rel_*(P,V)
* atomic_clear_acq_*(P,V)
* atomic_clear_rel_*(P,V)
* atomic_add_acq_*(P,V)
* atomic_add_rel_*(P,V)
* atomic_substract_acq_*(P,V)
* atomic_substract_rel_*(P,V)

以 atomic_cmpset_int(P,E, V) 為例，如果以 $C$/$C++$ 來實作，這個 op 將可能被發能兩個步驟：
# 比較 P 和 E 的內容，
# 若 P 和 E 的值相同，將 P 改為 V 值。
在多 CPU 旳環境下，P 的值可能在兩個步驟之間被改變，進而造成 V 蓋掉這個中間的值，使的中間值流失，造成邏輯上的錯誤。當有多個 CPU 同時要讀取，並更改同一變數時，我們會希望能依順序一個一個進行，而非不可預期的交錯其指令。為達成這個目的，使用不同平臺(CPU) ，會有不同的作法，因此 $FreeBSD$ 定義這些基本指令，由各個平臺各自 implement 。而 kernel 的其它部分，則基於基本指令，組合完成 CPU 同步的效果。在 i386 的實作中，atomic_cmpset_int(P,E,V) 以 inline assembly 實作如下:
{{{
#define MPLOCKED        lock ;

static __inline int                                                             
atomic_cmpset_int(volatile u_int *dst, u_int exp, u_int src)                    
{                                                                               
        int res = exp;                                                          
                                                                                
        __asm __volatile (                                                      
        "       lock                            "                                       
        "       cmpxchgl %2,%1 ;        "                                       
        "       setz    %%al ;          "                                       
        "       movzbl  %%al,%0 ;       "                                       
        "1:                             "                                       
        "# atomic_cmpset_int"                                                   
        : "+a" (res),                   /* 0 (result) */                        
          "=m" (*dst)                   /* 1 */                                 
        : "r" (src),                    /* 2 */                                 
          "m" (*dst)                    /* 3 */                                 
        : "memory");                                                            
                                                                                
        return (res);                                                           
}                                                                               
}}}
透過使用 lock 將 bus 佔有，然後完成 cmpxchgl 指令。這個指今本身分成讀取和寫入兩個 memory access 的動作，但是透過 lock 佔住 bus，完成 instruction level 的同步。lock 只對其後的第一個 CPU 指令生效，使其在執行期間，獨佔 system bus 。 i386 CPU 如此 implement ，而其它 CPU 會因為指令集的不同，使用不同的設計。 Kernel 透過這組 atomic op ，和 CPU decouple ，提供跨平臺的能力。

atomic_*_acq_*() 和 atomic_*_rel_*() 這些指令，實作 linkname:[acquire/release protocol] attach:WDDD05_praun.pdf 。引用:
* A release operation guarantees that updates to shared memory that the current process has issued prior to the release can be visible to other processors.
* An acquire operation guarantees that the current processor can correctly observe updates of shared memory that other processors have made available through a release operation (preceding the acquire).
acquire/release protocol 能解決 CPU cache 所造成的不同步，使的 cache 內的資料能及時的 update 到 core memory 。

$FreeBSD$ 似乎沒使用 relax memory barrier，以換取更高的效能。

=== Mutex ===
kern/kern_mutex.c 就是使用前述的 atomic op ，實作 linkname:mutex http://en.wikipedia.org/wiki/Mutex 的功能。 mutex 是 $FreeBSD$ kernel 使用的最基礎的 lock 功能，所有其它更複雜的 lock ，都是基於 mutex 以實作。

sys/_lock.h, kern/subr_look.c
{{{
struct lock_object {                                                            
        const   char *lo_name;          /* Individual lock name. */             
        const   char *lo_type;          /* General lock type. */                
        u_int   lo_flags;                                                       
        union {                         /* Data for witness. */                 
                STAILQ_ENTRY(lock_object) lod_list;                             
                struct  witness *lod_witness;                                   
        } lo_witness_data;                                                      
};                                                                              
}}}

sys/_mutex.h, kern/kern_mutex.c
{{{
struct mtx {                                                                    
        struct lock_object      mtx_object;     /* Common lock properties. */   
        volatile uintptr_t      mtx_lock;       /* Owner and flags. */          
        volatile u_int          mtx_recurse;    /* Number of recursive holds. *$$
                                                                                
#ifdef MUTEX_PROFILING                                                          
        /*                                                                      
         * This does not result in variant structure sizes because              
         * MUTEX_PROFILING is in opt_global.h                                   
         */                                                                     
        u_int64_t               mtx_acqtime;                                    
        const char              *mtx_filename;                                  
        int                     mtx_lineno;                                     
        /*                                                                      
         * Fields relating to measuring contention on mutexes.                  
         * holding must be accessed atomically since it's                       
         * modified by threads that don't yet hold the mutex.                   
         * locking is only modified and referenced while                        
         * the mutex is held.                                                   
         */                                                                     
        u_int                   mtx_contest_holding;                            
        u_int                   mtx_contest_locking;                            
#endif                                                                          
};                                                                              
}}}
使用前述的 atomic op ，更改 mtx_lock 欄位，以達成 lock 的目的。Mutex 以 mtx_lock 決定是否被 acquire，如果是，則 mtx_lock 將指向取得 mutex 的 thread structure 。

sys/mutex.h
* void mtx_init(struct mtx *m, const char *name, const char *type, int opts);
* void mtx_destroy(struct mtx *m);
* void mtx_sysinit(void *arg);
* void mutex_init(void);  # initialize system mutexes
* mtx_lock(m)
* mtx_lock_spin(m)
* mtx_trylock(m)
* mtx_unlock(m)
* mtx_unlock_spin(m)
* mtx_lock_flags(m, opts)
* mtx_unlock_flags(m, opts)
* mtx_lock_spin_flags(m, opts)
* mtx_unlock_spin_flags(m, opts)
* mtx_trylock_flags(m, opts)
* mtx_initialized(m)
* mtx_owned(m)
* mtx_recursed(m)
* mtx_name(m)
* mtx_assert(m, what)

使用$範例$:
{{{
struct mtx my_mtx;
char *data = NULL;

void thread_init(void) {
  mtx_init(&my_mtx, "myresource", NULL, MTX_DEF);
}

void thread_producer(void) {
  while(1) {
    mtx_lock(&my_mtx);
    if(data == NULL) {
      data = produce_data();
    }
    mtx_unlock(&my_mtx);
    mi_switch();  /* release CPU control */
  }
}

void thread_consumer(void) {
   while(1) {
    mtx_lock(&my_mtx);
    if(data != NULL) {
      consume_data(data);
      data = NULL;
    }
    mtx_unlock(&my_mtx);
    mi_switch();  /* release CPU control */
  }
}
}}}

=== lockmgr ==
lockmgr 實現一般常用的 lock 功能，如 share/exclusive lock 、 upgrade 、 downgrade 等。

sys/lockmgr.h
{{{
struct lock {                                                                   
        struct  mtx *lk_interlock;      /* lock on remaining fields */          
        u_int   lk_flags;               /* see below */                         
        int     lk_sharecount;          /* # of accepted shared locks */        
        int     lk_waitcount;           /* # of processes sleeping for lock */  
        short   lk_exclusivecount;      /* # of recursive exclusive locks */    
        short   lk_prio;                /* priority at which to sleep */        
        const char *lk_wmesg;           /* resource sleeping (for tsleep) */    
        int     lk_timo;                /* maximum sleep time (for tsleep) */   
        struct thread *lk_lockholder;   /* thread of exclusive lock holder */   
        struct  lock *lk_newlock;       /* lock taking over this lock */        
#ifdef  DEBUG_LOCKS                                                             
        struct stack lk_stack;                                                  
#endif                                                                          
};                                                                              
}}}
* int lockmgr(lkp, flags, interlkp, td)
* void transferlockers(from, to)
* void lockinit(lkp, prio, wmesg, timo, flags)
* void lockdestroy(lkp)
* int lockstatus(lkp, td)
* int lockcount(lkp)
* int lockwaiters(lkp)

最後更新時間: 2006-11-15 11:53:18 CST | 引用

查詢:

COMMENTS: