Chinaunix首页 | 论坛 | 博客
  • 博客访问: 2402027
  • 博文数量: 862
  • 博客积分: 14125
  • 博客等级: 上将
  • 技术积分: 10598
  • 用 户 组: 普通用户
  • 注册时间: 2007-07-27 16:53
个人简介

https://github.com/zytc2009/BigTeam_learning

文章分类

全部博文(862)

文章存档

2019年(3)

2018年(1)

2017年(10)

2015年(3)

2014年(8)

2013年(3)

2012年(69)

2011年(103)

2010年(357)

2009年(283)

2008年(22)

分类: LINUX

2010-11-13 16:44:06

semaphore是内核中比较重要和常用的同步方式之一,他主要的特点是实现了Sleep机制下的同步。也就是当获取 一个semaphore但是又不能立刻获取的时候,他使当前的执行进程进入到Sleep状态中等待,当semaphore可以获取的时候,从新开始运行, 而不像splin lock在获取锁的时候是BusyWait。

首先看其定义:

struct semaphore {
                atomic_t count; // 原子变量,是后续的实际代码中,我们能看到其即为我们在初始化时所设置的信号量。
                int sleepers; // 有几个等待者。
                wait_queue_head_t wait; // 等待队列
        };

初始化函数:
        static inline void sema_init (struct semaphore *sem, int val)
        {
                atomic_set(&sem->count, val); // 原子操作,把信号量的值设为原子操作的值。
                sem->sleepers = 0; // 设为等待者为0。
                init_waitqueue_head(&sem->wait); // 初始化等待队列。
        }

好看完了初始化函数,我们来看一下一个特例PV操作:

static inline void init_MUTEX (struct semaphore *sem)
        {
                sema_init(sem, 1);
        }

static inline void init_MUTEX_LOCKED (struct semaphore *sem)
        {
                sema_init(sem, 0);
        }

从中我们可以看到,其实我们所说的PV操作就是调用sema_init来把其中的原子变量分别置0或者1。

下面我们来看下具体的操作函数:

static inline void down(struct semaphore * sem)
        {
                might_sleep();
                __asm__ __volatile__(
                "# atomic down operation\n\t"
                LOCK_PREFIX "decl %0\n\t" /* --sem->count */
                "jns 2f\n"
                "\tlea %0,%%eax\n\t"
                "call __down_failed\n"
                "2:"
                :"+m" (sem->count)
                :
                :"memory","ax");
        }

关于might_sleep():
        #define might_sleep() \
                do { __might_sleep(__FILE__, __LINE__); might_resched(); } while (0)

从中可以看到其根本是调用might_resched():
        #define might_resched() cond_resched()

其调用了cond_resched():

int __sched cond_resched(void)
        {
                if (need_resched() && !(preempt_count() & PREEMPT_ACTIVE) &&
                system_state == SYSTEM_RUNNING) {
                        __cond_resched();
                        return 1;
                 }
                return 0;
        }

其中:

static inline int need_resched(void)
        {
                return unlikely(test_thread_flag(TIF_NEED_RESCHED));
                // TIF_NEED_RESCHED 这个值在arm是2,在i386中是3。
        }

其是判断进程的状态是否是需要调度。

其具体实现是:

#define test_thread_flag(flag) \
        test_ti_thread_flag(current_thread_info(), flag)

得到当前进程的状态,然后和flag做位&操作,判断是否可以做调度。

看下关于抢占机制的判断:
        #define preempt_count() (current_thread_info()->preempt_count)

调用preempt_count()得到当前进程关于抢占的状态,preempt_count & PREEMPT_ACTIVE 为TRUE,取反为当前不可以抢占,就不能调度,否则为可调度。

看些系统状态:system_state 为运行状态。

综合上面的几个条件,我们看出当前进程可以被调度或者说抢占,于是当前进程被调度放弃CPU资源。
might_sleep宏就是检查是否需要重新调度。

下面我们进入汇编代码区:

__asm__ __volatile__(
        "# atomic up operation\n\t"
        LOCK_PREFIX "decl %0\n\t" /* --sem->count */
        "jns 2f\n"        /* 当符号标志位为0的时候跳转到标号2地方,标志获取锁,结束本次操作。 */
        "\tlea %0,%%eax\n\t"
        "call __down_failed\n"        /* 获取不到的话,压栈,调用 __down_failed */
        "2:"
        :"+m" (sem->count)
        :
        :"memory","ax");

LOCK_PREFIX宏:

在多处理器环境中 LOCK_PREFIX 实际被定义为 “lock”前缀。

x86 处理器使用“lock”前缀的方式提供了在指令执行期间对总线加锁的手段。芯片上有一条引线 LOCK,如果在一条汇编指令(ADD, ADC, AND, BTC, BTR, BTS, CMPXCHG, CMPXCH8B, DEC, INC, NEG, NOT, OR, SBB, SUB, XOR, XADD, XCHG)前加上“lock” 前缀,经过汇编后的机器代码就使得处理器执行该指令时把引线 LOCK 的电位拉低,从而把总线锁住,这样其它处理器或使用DMA的外设暂时无法通过同一总线访问内存。

注意:从 P6 处理器开始,如果指令访问的内存区域已经存在于处理器的内部缓存中,则“lock” 前缀并不将引线 LOCK 的电位拉低,而是锁住本处理器的内部缓存,然后依靠缓存一致性协议保证操作的原子性。

在比较新的内核版本中Version >2.6.20, i386平台找不到关于__down_failed的具体定义:

linux-2.6.13.4>arch>i386>kernel>semaphore.c
        asm(
        ".section .sched.text\n"
        ".align 4\n"
        ".globl __down_failed\n"
        "__down_failed:\n\t"
        #if defined(CONFIG_FRAME_POINTER)
                "pushl %ebp\n\t"
                "movl %esp,%ebp\n\t"
        #endif
                "pushl %edx\n\t"
                "pushl %ecx\n\t"
                "call __down\n\t"
                "popl %ecx\n\t"
                "popl %edx\n\t"
        #if defined(CONFIG_FRAME_POINTER)
                "movl %ebp,%esp\n\t"
                "popl %ebp\n\t"
        #endif
                "ret"
        );

从汇编代码中我们可以看到其调用了__down()来实现其等待获取锁的过程。

fastcall void __sched __down(struct semaphore * sem)
        {
                struct task_struct *tsk = current; // 当前进程task_struct。
                DECLARE_WAITQUEUE(wait, tsk); // 定义等待队列。
                unsigned long flags;

        tsk->state = TASK_UNINTERRUPTIBLE; // 设置当前进程状态为不可中断状态。
                spin_lock_irqsave(&sem->wait.lock, flags);
                // spin_loc_irqsave 在获得自旋锁之前禁止中断(只在当前处理器);之前的中断状态保存在 flags 里。
                add_wait_queue_exclusive_locked(&sem->wait, &wait);
                        // 将当前的进程添加到等待队列的队尾。操作是不需要锁,因为此函数一般是在获取锁之后调用。

        sem->sleepers++;
                for (;;) {
                        int sleepers = sem->sleepers;

                /*
                        * Add "everybody else" into it. They aren't
                        * playing, because we own the spinlock in
                        * the wait_queue_head.
                        */
                        if (!atomic_add_negative(sleepers - 1, &sem->count)) {
                                        // 将(sleepers - 1) + ( &sem->count),如果当结果值为负数时返回真。
                                        /* First:初始化的时候,count = val, sleeper = 0,
                                        * down()中, count -- 成为 val - 1,以PV操作为例,其val值为1,count = 0,
                                        * __down()中,sleeper ++ 为 1,
                                        * 此时(sleepers - 1, &sem->count) 为0,不为负值,故此为FALSE,跳出循环。
                                        *
                                        * Two:由于在跳出的时候sleepers设置为0,故此Other进程尝试想再次获取锁资源的时候,
                                        * (sleepers - 1, &sem->count) 为 ( 0 - 1, 0 )为负值,此时为真,进入下次循环。
                                        */
                               sem->sleepers = 0; //
                               break;
                       }
                       sem->sleepers = 1;        /* us - see -1 above */
                                                                     /* 设置sleeper为1,标识有进程在等待。*/
                       spin_unlock_irqrestore(&sem->wait.lock, flags);
                                                                     /* 解开锁,开始准备被调度。 */

                schedule();

                spin_lock_irqsave(&sem->wait.lock, flags);
                        tsk->state = TASK_UNINTERRUPTIBLE;
                }
                remove_wait_queue_locked(&sem->wait, &wait);
                                /* 获取到了锁资源,把当前进程从等待队列中删掉。 */
                wake_up_locked(&sem->wait);
                                /* 唤醒等待队列 */
                spin_unlock_irqrestore(&sem->wait.lock, flags);
                tsk->state = TASK_RUNNING;
        }

这里要注意的是对sleeper成员的理解结合后面的代码:

进程在没有获得锁时会将其设为1,在获得了锁的时候将其设为0。

所以,sleeper = 1时表示有进程在等待,sleeper = 0时,有进程获得信号量退出,或者没有进程在等待此信号量。

如果在等待队列中进程被唤醒并获得了锁,则将sleeper设为0,然后用break退出循环,再用wake_up_locked(&sem->wait)唤醒等待队列中的一个进程。
这个进程下次被调度的时候从shedule()后面开始运行,更改进程状态。

调用if (!atomic_add_negative(sleepers - 1, &sem->count))判断,此时sleeper = 0,sleep - 1 = -1,经过atomic_add_negative()使sem->count - 1了。

因此前面论述过,进程进来取信号量 down()里count有减1,然后如果有等待线程在获取到这个信号,刚又会将count + 1,使其变到了原值。

现在在我们这个情景中,已经有线程释放信号了,应该要 + 1。

那思考一下:

为什么要对count sleeper做这样的处理呢?直接使用使用计数不就完了吗?进程等待获取信号量的时候sleep+1 count-1.释放信号量的时候sleep-1.count+1不可以吗?

表面上上述的方法也能工作的很好,但是如果进程数目一多就要考虑到溢出问题了。

static inline void up(struct semaphore * sem)
        {
        __asm__ __volatile__(
                "# atomic up operation\n\t"
                LOCK_PREFIX "incl %0\n\t" /* ++sem->count */
                "jg 1f\n\t"
                "lea %0,%%eax\n\t"
                "call __up_wakeup\n"
                "1:"
                :"+m" (sem->count)
                :
                :"memory","ax");
        }

        asm(
        ".section .sched.text\n"
        ".align 4\n"
        ".globl __up_wakeup\n"
        "__up_wakeup:\n\t"
                "pushl %edx\n\t"
                "pushl %ecx\n\t"
                "call __up\n\t"
                "popl %ecx\n\t"
                "popl %edx\n\t"
                "ret"
        );

fastcall void __up(struct semaphore *sem)
        {
                wake_up(&sem->wait);
        }

释放锁资源就比较简单了,首先是利用汇编判断其是否是有符号大于,我们这里count是为0,不是大于则call __up_wakeup(),最终执行__up(),唤醒等待队列。

至于相应信号的分析如下:

static inline int down_interruptible(struct semaphore * sem)
        {
                int result;

        might_sleep();
                __asm__ __volatile__(
                        "# atomic interruptible down operation\n\t"
                        "xorl %0,%0\n\t"
                        LOCK_PREFIX "decl %1\n\t"        /* --sem->count */
                        "jns 2f\n\t"
                        "lea %1,%%eax\n\t"
                        "call __down_failed_interruptible\n"
                        "2:"
                        :"=&a" (result), "+m" (sem->count)
:
                        :"memory");
                return result;
        }

fastcall int __sched __down_interruptible(struct semaphore * sem)
        {
                int retval = 0; /* 返回值 */
                struct task_struct *tsk = current;
                DECLARE_WAITQUEUE(wait, tsk);
                unsigned long flags;

        tsk->state = TASK_INTERRUPTIBLE;
                spin_lock_irqsave(&sem->wait.lock, flags);
                add_wait_queue_exclusive_locked(&sem->wait, &wait);

        sem->sleepers++;
                for (;;) {
                        int sleepers = sem->sleepers;

                /*
                        * With signals pending, this turns into
                        * the trylock failure case - we won't be
                        * sleeping, and we* can't get the lock as
                        * it has contention. Just correct the count
                        * and exit.
                        */
                        if (signal_pending(current)) {
                                /* 判断是否有中断触发,如果有则返回非0值,否则返回0。*/
                                retval = -EINTR;
                                sem->sleepers = 0;
                                /* 重置sleepers = 0 */
                                atomic_add(sleepers, &sem->count);
                                /* 让等待者等于count。*/
                                break;
                        }

                /*
                        * Add "everybody else" into it. They aren't
                        * playing, because we own the spinlock in
                        * wait_queue_head. The "-1" is because we're
                        * still hoping to get the semaphore.
                        */
                        if (!atomic_add_negative(sleepers - 1, &sem->count)) {
                                sem->sleepers = 0;
                                break;
                        }
                        sem->sleepers = 1; /* us - see -1 above */
                        spin_unlock_irqrestore(&sem->wait.lock, flags);

                schedule();

                spin_lock_irqsave(&sem->wait.lock, flags);
                        tsk->state = TASK_INTERRUPTIBLE;
                }
                remove_wait_queue_locked(&sem->wait, &wait);
                wake_up_locked(&sem->wait);
                spin_unlock_irqrestore(&sem->wait.lock, flags);

        tsk->state = TASK_RUNNING;
                return retval;
        }

static inline int down_trylock(struct semaphore * sem)
       {
                  int result;

         __asm__ __volatile__(
                           "# atomic interruptible down operation\n\t"
                           "xorl %0,%0\n\t"
                           LOCK_PREFIX "decl %1\n\t" /* --sem->count */
                           "jns 2f\n\t" /* 可以获取锁,锁之。*/
                           "lea %1,%%eax\n\t"
                           "call __down_failed_trylock\n\t"
                           "2:\n"
                           :"=&a" (result), "+m" (sem->count)
                           :
                           :"memory");
                return result;
        }

fastcall int __down_trylock(struct semaphore * sem)
        {
                int sleepers;
                unsigned long flags;

        spin_lock_irqsave(&sem->wait.lock, flags);
                sleepers = sem->sleepers + 1;
                        /* 不能获取锁,把sleepers + 1,判断是否有等待进程,如果有唤醒其。*/
                sem->sleepers = 0;

        /*
                * Add "everybody else" and us into it. They aren't
                * playing, because we own the spinlock in the
                * wait_queue_head.
                */
                if (!atomic_add_negative(sleepers, &sem->count)) {
                wake_up_locked(&sem->wait);
                }

        spin_unlock_irqrestore(&sem->wait.lock, flags);
                return 1;
        }
阅读(535) | 评论(0) | 转发(0) |
给主人留下些什么吧!~~