pthread |
您所在的位置：网站首页 › pthread_cond_init源码解析 › pthread |
pthread
|
1.Linux“线程”
进程与线程之间是有区别的，不过linux内核只提供了轻量进程的支持，未实现线程模型。Linux是一种“多进程单线程”的操作系统。Linux本身只有进程的概念，而其所谓的“线程”本质上在内核里仍然是进程。
大家知道，进程是资源分配的单位，同一进程中的多个线程共享该进程的资源（如作为共享内存的全局变量）。Linux中所谓的“线程”只是在被创建时clone了父进程的资源，因此clone出来的进程表现为“线程”，这一点一定要弄清楚。因此，Linux“线程”这个概念只有在打冒号的情况下才是最准确的。
目前Linux中最流行的线程机制为LinuxThreads，所采用的就是线程－进程“一对一”模型，调度交给核心，而在用户级实现一个包括信号处理在内的线程管理机制。LinuxThreads由Xavier Leroy ([email protected])负责开发完成，并已绑定在GLIBC中发行，它实现了一种BiCapitalized面向Linux的Posix 1003.1c “pthread”标准接口。Linuxthread可以支持Intel、Alpha、MIPS等平台上的多处理器系统。
按照POSIX 1003.1c 标准编写的程序与Linuxthread 库相链接即可支持Linux平台上的多线程，在程序中需包含头文件pthread. h，在编译链接时使用命令： gcc -D -REENTRANT -lpthread xxx. c其中-REENTRANT宏使得相关库函数(如stdio.h、errno.h中函数) 是可重入的、线程安全的(thread-safe)，-lpthread则意味着链接库目录下的libpthread.a或libpthread.so文件。使用Linuxthread库需要2.0以上版本的Linux内核及相应版本的C库(libc 5.2.18、libc 5.4.12、libc 6)。 2.“线程”控制线程创建 进程被创建时，系统会为其创建一个主线程，而要在进程中创建新的线程，则可以调用pthread_create： pthread_create(pthread_t *thread, const pthread_attr_t *attr, void * (start_routine)(void*), void *arg);start_routine为新线程的入口函数，arg为传递给start_routine的参数。 每个线程都有自己的线程ID，以便在进程内区分。线程ID在pthread_create调用时回返给创建线程的调用者；一个线程也可以在创建后使用pthread_self()调用获取自己的线程ID： pthread_self (void) ;线程退出 线程的退出方式有三： （1）执行完成后隐式退出； （2）由线程本身显示调用pthread_exit 函数退出； pthread_exit (void * retval) ;（3）被其他线程用pthread_cance函数终止： pthread_cance (pthread_t thread) ;在某线程中调用此函数，可以终止由参数thread 指定的线程。 如果一个线程要等待另一个线程的终止，可以使用pthread_join函数，该函数的作用是调用pthread_join的线程将被挂起直到线程ID为参数thread的线程终止： pthread_join (pthread_t thread, void** threadreturn); 3.线程通信线程互斥 互斥意味着“排它”，即两个线程不能同时进入被互斥保护的代码。Linux下可以通过pthread_mutex_t 定义互斥体机制完成多线程的互斥操作，该机制的作用是对某个需要互斥的部分，在进入时先得到互斥体，如果没有得到互斥体，表明互斥部分被其它线程拥有，此时欲获取互斥体的线程阻塞，直到拥有该互斥体的线程完成互斥部分的操作为止。 下面的代码实现了对共享全局变量x 用互斥体mutex 进行保护的目的： int x; // 进程中的全局变量 pthread_mutex_t mutex; pthread_mutex_init(&mutex, NULL); //按缺省的属性初始化互斥体变量mutex pthread_mutex_lock(&mutex); // 给互斥体变量加锁 … //对变量x 的操作 phtread_mutex_unlock(&mutex); // 给互斥体变量解除锁线程同步 同步就是线程等待某个事件的发生。只有当等待的事件发生线程才继续执行，否则线程挂起并放弃处理器。当多个线程协作时，相互作用的任务必须在一定的条件下同步。 Linux下的C语言编程有多种线程同步机制，最典型的是条件变量(condition variable)。pthread_cond_init用来创建一个条件变量，其函数原型为： pthread_cond_init (pthread_cond_t *cond, const pthread_condattr_t *attr);pthread_cond_wait和pthread_cond_timedwait用来等待条件变量被设置，值得注意的是这两个等待调用需要一个已经上锁的互斥体mutex，这是为了防止在真正进入等待状态之前别的线程有可能设置该条件变量而产生竞争。pthread_cond_wait的函数原型为： pthread_cond_wait (pthread_cond_t *cond, pthread_mutex_t *mutex);pthread_cond_broadcast用于设置条件变量，即使得事件发生，这样等待该事件的线程将不再阻塞： pthread_cond_broadcast (pthread_cond_t *cond) ;pthread_cond_signal则用于解除某一个等待线程的阻塞状态： pthread_cond_signal (pthread_cond_t *cond) ; pthread_cond_destroy 则用于释放一个条件变量的资源。在头文件semaphore.h 中定义的信号量则完成了互斥体和条件变量的封装，按照多线程程序设计中访问控制机制，控制对资源的同步访问，提供程序设计人员更方便的调用接口。 sem_init(sem_t *sem, int pshared, unsigned int val);这个函数初始化一个信号量sem 的值为val，参数pshared 是共享属性控制，表明是否在进程间共享。 sem_wait(sem_t *sem);调用该函数时，若sem为无状态，调用线程阻塞，等待信号量sem值增加(post )成为有信号状态；若sem为有状态，调用线程顺序执行，但信号量的值减一。 sem_post(sem_t *sem);调用该函数，信号量sem的值增加，可以从无信号状态变为有信号状态。 4.实例下面我们还是以名的生产者/消费者问题为例来阐述Linux线程的控制和通信。一组生产者线程与一组消费者线程通过缓冲区发生联系。生产者线程将生产的产品送入缓冲区，消费者线程则从中取出产品。缓冲区有N 个，是一个环形的缓冲池。 #include #include #define BUFFER_SIZE 16 // 缓冲区数量 struct prodcons { // 缓冲区相关数据结构 int buffer[BUFFER_SIZE]; /* 实际数据存放的数组*/ pthread_mutex_t lock; /* 互斥体lock 用于对缓冲区的互斥操作 */ int readpos, writepos; /* 读写指针*/ pthread_cond_t notempty; /* 缓冲区非空的条件变量 */ pthread_cond_t notfull; /* 缓冲区未满的条件变量 */ }; /* 初始化缓冲区结构 */ void init(struct prodcons *b) { pthread_mutex_init(&b->lock, NULL); pthread_cond_init(&b->notempty, NULL); pthread_cond_init(&b->notfull, NULL); b->readpos = 0; b->writepos = 0; } /* 将产品放入缓冲区,这里是存入一个整数*/ void put(struct prodcons *b, int data) { pthread_mutex_lock(&b->lock); /* 等待缓冲区未满*/ if ((b->writepos + 1) % BUFFER_SIZE == b->readpos) { pthread_cond_wait(&b->notfull, &b->lock); } /* 写数据,并移动指针 */ b->buffer[b->writepos] = data; b->writepos++; if (b->writepos >= BUFFER_SIZE) b->writepos = 0; /* 设置缓冲区非空的条件变量*/ pthread_cond_signal(&b->notempty); pthread_mutex_unlock(&b->lock); } /* 从缓冲区中取出整数*/ int get(struct prodcons *b) { int data; pthread_mutex_lock(&b->lock); /* 等待缓冲区非空*/ if (b->writepos == b->readpos) { pthread_cond_wait(&b->notempty, &b->lock); } /* 读数据,移动读指针*/ data = b->buffer[b->readpos]; b->readpos++; if (b->readpos >= BUFFER_SIZE) b->readpos = 0; /* 设置缓冲区未满的条件变量*/ pthread_cond_signal(&b->notfull); pthread_mutex_unlock(&b->lock); return data; } /* 测试:生产者线程将1 到10000 的整数送入缓冲区,消费者线 程从缓冲区中获取整数,两者都打印信息*/ #define OVER ( - 1) struct prodcons buffer; void *producer(void *data) { int n; for (n = 0; n < 10000; n++) { printf("%d --->\n", n); put(&buffer, n); } put(&buffer, OVER); return NULL; } void *consumer(void *data) { int d; while (1) { d = get(&buffer); if (d == OVER) break; printf("--->%d \n", d); } return NULL; } int main(void) { pthread_t th_a, th_b; void *retval; init(&buffer); /* 创建生产者和消费者线程*/ pthread_create(&th_a, NULL, producer, 0); pthread_create(&th_b, NULL, consumer, 0); /* 等待两个线程结束*/ pthread_join(th_a, &retval); pthread_join(th_b, &retval); return 0; } 5.WIN32、VxWorks、Linux线程类比目前为止，笔者已经创作了《基于嵌入式操作系统VxWorks的多任务并发程序设计》（《软件报》2006年5~12期连载）、《深入浅出Win32多线程程序设计》（天极网技术专题）系列，我们来找出这两个系列文章与本文的共通点。 看待技术问题要瞄准其本质，不管是Linux、VxWorks还是WIN32，其涉及到多线程的部分都是那些内容，无非就是线程控制和线程通信，它们的许多函数只是名称不同，其实质含义是等价的，下面我们来列个三大操作系统共同点详细表单：
本章讲述了Linux下多线程的控制及线程间通信编程方法，给出了一个生产者/消费者的实例，并将Linux的多线程与WIN32、VxWorks多线程进行了类比，总结了一般规律。鉴于多线程编程已成为开发并发应用程序的主流方法，学好本章的意义也便不言自明。 完 #include #include #include void thread(void) { int i; for(i=0;i__data.__wrefs); clockid_t clockid = (flags & __PTHREAD_COND_CLOCK_MONOTONIC_MASK) ? CLOCK_MONOTONIC : CLOCK_REALTIME; return __pthread_cond_wait_common (cond, mutex, clockid, abstime); } /* This condvar implementation guarantees that all calls to signal and broadcast and all of the three virtually atomic parts of each call to wait (i.e., (1) releasing the mutex and blocking, (2) unblocking, and (3) re- acquiring the mutex) happen in some total order that is consistent with the happens-before relations in the calling program. However, this order does not necessarily result in additional happens-before relations being established (which aligns well with spurious wake-ups being allowed). All waiters acquire a certain position in a 64b waiter sequence (__wseq). This sequence determines which waiters are allowed to consume signals. A broadcast is equal to sending as many signals as are unblocked waiters. When a signal arrives, it samples the current value of __wseq with a relaxed-MO load (i.e., the position the next waiter would get). (This is sufficient because it is consistent with happens-before; the caller can enforce stronger ordering constraints by calling signal while holding the mutex.) Only waiters with a position less than the __wseq value observed by the signal are eligible to consume this signal. This would be straight-forward to implement if waiters would just spin but we need to let them block using futexes. Futexes give no guarantee of waking in FIFO order, so we cannot reliably wake eligible waiters if we just use a single futex. Also, futex words are 32b in size, but we need to distinguish more than 1> 1)) goto done; /* TODO Back off. */ /* Reload signals. See above for MO. */ signals = atomic_load_acquire (cond->__data.__g_signals + g); spin--; } /* If our group will be closed as indicated by the flag on signals, don't bother grabbing a signal. */ if (signals & 1) goto done; /* If there is an available signal, don't block. */ if (signals != 0) break; /* No signals available after spinning, so prepare to block. We first acquire a group reference and use acquire MO for that so that we synchronize with the dummy read-modify-write in __condvar_quiesce_and_switch_g1 if we read from that. In turn, in this case this will make us see the closed flag on __g_signals that designates a concurrent attempt to reuse the group's slot. We use acquire MO for the __g_signals check to make the __g1_start check work (see spinning above). Note that the group reference acquisition will not mask the release MO when decrementing the reference count because we use an atomic read-modify-write operation and thus extend the release sequence. */ atomic_fetch_add_acquire (cond->__data.__g_refs + g, 2); if (((atomic_load_acquire (cond->__data.__g_signals + g) & 1) != 0) || (seq < (__condvar_load_g1_start_relaxed (cond) >> 1))) { /* Our group is closed. Wake up any signalers that might be waiting. */ __condvar_dec_grefs (cond, g, private); goto done; } // Now block. struct _pthread_cleanup_buffer buffer; struct _condvar_cleanup_buffer cbuffer; cbuffer.wseq = wseq; cbuffer.cond = cond; cbuffer.mutex = mutex; cbuffer.private = private; __pthread_cleanup_push (&buffer, __condvar_cleanup_waiting, &cbuffer); if (abstime == NULL) { /* Block without a timeout. */ err = futex_wait_cancelable ( cond->__data.__g_signals + g, 0, private); } else { /* Block, but with a timeout. Work around the fact that the kernel rejects negative timeout values despite them being valid. */ if (__glibc_unlikely (abstime->tv_sec < 0)) err = ETIMEDOUT; else { err = futex_abstimed_wait_cancelable (cond->__data.__g_signals + g, 0, clockid, abstime, private); } } __pthread_cleanup_pop (&buffer, 0); if (__glibc_unlikely (err == ETIMEDOUT)) { __condvar_dec_grefs (cond, g, private); /* If we timed out, we effectively cancel waiting. Note that we have decremented __g_refs before cancellation, so that a deadlock between waiting for quiescence of our group in __condvar_quiesce_and_switch_g1 and us trying to acquire the lock during cancellation is not possible. */ __condvar_cancel_waiting (cond, seq, g, private); result = ETIMEDOUT; goto done; } else __condvar_dec_grefs (cond, g, private); /* Reload signals. See above for MO. */ signals = atomic_load_acquire (cond->__data.__g_signals + g); } } /* Try to grab a signal. Use acquire MO so that we see an up-to-date value of __g1_start below (see spinning above for a similar case). In particular, if we steal from a more recent group, we will also see a more recent __g1_start below. */ while (!atomic_compare_exchange_weak_acquire (cond->__data.__g_signals + g, &signals, signals - 2)); /* We consumed a signal but we could have consumed from a more recent group that aliased with ours due to being in the same group slot. If this might be the case our group must be closed as visible through __g1_start. */ uint64_t g1_start = __condvar_load_g1_start_relaxed (cond); if (seq < (g1_start >> 1)) { /* We potentially stole a signal from a more recent group but we do not know which group we really consumed from. We do not care about groups older than current G1 because they are closed; we could have stolen from these, but then we just add a spurious wake-up for the current groups. We will never steal a signal from current G2 that was really intended for G2 because G2 never receives signals (until it becomes G1). We could have stolen a signal from G2 that was conservatively added by a previous waiter that also thought it stole a signal -- but given that that signal was added unnecessarily, it's not a problem if we steal it. Thus, the remaining case is that we could have stolen from the current G1, where "current" means the __g1_start value we observed. However, if the current G1 does not have the same slot index as we do, we did not steal from it and do not need to undo that. This is the reason for putting a bit with G2's index into__g1_start as well. */ if (((g1_start & 1) ^ 1) == g) { /* We have to conservatively undo our potential mistake of stealing a signal. We can stop trying to do that when the current G1 changes because other spinning waiters will notice this too and __condvar_quiesce_and_switch_g1 has checked that there are no futex waiters anymore before switching G1. Relaxed MO is fine for the __g1_start load because we need to merely be able to observe this fact and not have to observe something else as well. ??? Would it help to spin for a little while to see whether the current G1 gets closed? This might be worthwhile if the group is small or close to being closed. */ unsigned int s = atomic_load_relaxed (cond->__data.__g_signals + g); while (__condvar_load_g1_start_relaxed (cond) == g1_start) { /* Try to add a signal. We don't need to acquire the lock because at worst we can cause a spurious wake-up. If the group is in the process of being closed (LSB is true), this has an effect similar to us adding a signal. */ if (((s & 1) != 0) || atomic_compare_exchange_weak_relaxed (cond->__data.__g_signals + g, &s, s + 2)) { /* If we added a signal, we also need to add a wake-up on the futex. We also need to do that if we skipped adding a signal because the group is being closed because while __condvar_quiesce_and_switch_g1 could have closed the group, it might stil be waiting for futex waiters to leave (and one of those waiters might be the one we stole the signal from, which cause it to block using the futex). */ futex_wake (cond->__data.__g_signals + g, 1, private); break; } /* TODO Back off. */ } } } done: /* Confirm that we have been woken. We do that before acquiring the mutex to allow for execution of pthread_cond_destroy while having acquired the mutex. */ __condvar_confirm_wakeup (cond, private); /* Woken up; now re-acquire the mutex. If this doesn't fail, return RESULT, which is set to ETIMEDOUT if a timeout occured, or zero otherwise. */ err = __pthread_mutex_cond_lock (mutex); /* XXX Abort on errors that are disallowed by POSIX? */ return (err != 0) ? err : result; }源码比较难懂 但可以看见有调用 err = __pthread_mutex_unlock_usercnt (mutex, 0); ... err = __pthread_mutex_cond_lock (mutex); 应该是调用pthread_cond_wait后先释放锁，然后加入waiter队列等待signal，如果有signal唤醒了该线程则后续执行重新上锁 |
【本文地址】
今日新闻 |
推荐新闻 |