随笔 - 171  文章 - 257  trackbacks - 0
<2007年2月>
28293031123
45678910
11121314151617
18192021222324
25262728123
45678910

常用链接

留言簿(33)

随笔分类(225)

随笔档案(171)

相册

技术

友情链接

最新随笔

搜索

  •  

积分与排名

  • 积分 - 440592
  • 排名 - 48

最新随笔

最新评论

阅读排行榜

评论排行榜

进行多线程编程,最头疼的就是那些共享的数据。因为你无法知道哪个线程会在哪个时候对它进行操作,你也无法得知那个线程会先运行,哪个线程会后运行。下面介绍一些技术,通过他们,你会合理安排你的线程之间对资源的竞争。

l         互斥体Mutex

l         信号灯Semophore

l         条件变量Conditions

先说一下互斥量。

什么时候会用上互斥量了?比如你现在有一全局链表,你有几个工作线程。每一个线程从该链表中取出头节点,然后对该头节点进行处理。比如现在线程1正在取出头节点,他的操作如下:

Item * p =queue_list;

Queue_list=queue_list->next;

Process_job(p);

Free(p);

当线程1处理完第一步,也就是Item *p=queue_list后,这时候系统停止线程1的运行,改而运行线程2。线程2照样取出头节点,然后进行处理,最后释放了该节点。过了段时间,线程1重新得到运行。而这个时候,其实p所指向的节点已经被线程2释放掉,而线程1对此毫无知晓。他会接着运行process_job(p)。而这将导致无法预料的后果!

对于这种情况,系统给我们提供了互斥量。你在取出头节点前必须要等待互斥量,如果此时有其他线程已经获得该互斥量,那么线程将会阻塞在这个地方。只有等到其他线程释放掉该互斥量后,你的线程才有可能得到该互斥量。为什么是可能了?因为可能此时有不止你一个线程在等候该互斥量,而系统无法保证你的线程将会优先运行。

互斥量的类型为pthread_mutex_t。你可以声明多个互斥量。在声明该变量后,你需要调用pthread_mutex_init()来创建该变量。pthread_mutex_init的格式如下:

int  pthread_mutex_init(pthread_mutex_t  *mutex,  const  pthread_mutex_attr_t *mutexattr);

第一个参数,mutext,也就是你之前声明的那个互斥量,第二个参数为该互斥量的属性。这个将在后面详细讨论。

在创建该互斥量之后,你便可以使用它了。要得到互斥量,你需要调用下面的函数:

int pthread_mutex_lock(pthread_mutex_t *mutex);

该函数用来给互斥量上锁,也就是我们前面所说的等待操作。互斥量一旦被上锁后,其他线程如果想给该互斥量上锁,那么就会阻塞在这个操作上。如果在此之前该互斥量已经被其他线程上锁,那么该操作将会一直阻塞在这个地方,直到获得该锁为止。

在得到互斥量后,你就可以进入关键代码区了。

同样,在操作完成后,你必须调用下面的函数来给互斥量解锁,也就是前面所说的释放。这样其他等待该锁的线程才有机会获得该锁,否则其他线程将会永远阻塞。

int pthread_mutex_unlock(pthread_mutex_t *mutex);

下面给出一个简单的例子:

#include <malloc.h>

#include <pthread.h>

struct job {

/* Link field for linked list. */

struct job* next;

/* Other fields describing work to be done... */

};

/* A linked list of pending jobs. */

struct job* job_queue;

/* A mutex protecting job_queue. */

pthread_mutex_t job_queue_mutex = PTHREAD_MUTEX_INITIALIZER;

/* Process queued jobs until the queue is empty. */

void* thread_function (void* arg)

{

while (1) {

struct job* next_job;

/* Lock the mutex on the job queue. */

pthread_mutex_lock (&job_queue_mutex);

/* Now it’s safe to check if the queue is empty. */

if (job_queue == NULL)

next_job = NULL;

else {

/* Get the next available job. */

next_job = job_queue;

/* Remove this job from the list. */

job_queue = job_queue->next;

}

/* Unlock the mutex on the job queue because we’re done with the

queue for now. */

pthread_mutex_unlock (&job_queue_mutex);

/* Was the queue empty? If so, end the thread. */

if (next_job == NULL)

break;

/* Carry out the work. */

process_job (next_job);

/* Clean up. */

free (next_job);

}

return NULL;

}

 

在这个例子中我们使用了下面一条语句:

pthread_mutex_t job_queue_mutex = PTHREAD_MUTEX_INITIALIZER;

       他的作用和调用pthread_mutex_init()函数一样。

 

如果一个线程已经给一个互斥量上锁了,后来在操作的过程中又再次调用了该上锁的操作,那么该线程将会无限阻塞在这个地方,从而导致死锁。怎么变了?这就需要我们之前所提到的互斥量的属性。

互斥量分为下面三种:

l         快速型。这种类型也是默认的类型。该线程的行为正如上面所说的。

l         递归型。如果遇到我们上面所提到的死锁情况,同一线程循环给互斥量上锁,那么系统将会知道该上锁行为来自同一线程,那么就会同意线程给该互斥量上锁。

l         错误检测型。如果该互斥量已经被上锁,那么后续的上锁将会失败而不会阻塞,pthread_mutex_lock()操作将会返回EDEADLK。

互斥量的属性类型为pthread_mutexattr_t。声明后调用pthread_mutexattr_init()来创建该互斥量。然后调用int pthread_mutexattr_settype(pthread_mutexattr_t *attr, int kind);来设置属性。int pthread_mutexattr_settype(pthread_mutexattr_t *attr, int kind);格式如下:

int pthread_mutexattr_settype(pthread_mutexattr_t *attr, int kind);

第一个参数,attr,就是前面声明的属性变量,第二个参数,kind,就是我们要设置的属性类型。他有下面几个选项:

l         PTHREAD_MUTEX_FAST_NP

l         PTHREAD_MUTEX_RECURSIVE_NP

l         PTHREAD_MUTEX_ERRORCHECK_NP

下面给出一个使用属性的简单过程:

pthread_mutex_t mutex;

pthread_mutexattr_t attr;

pthread_mutexattr_init(&attr);

pthread_mutexattr_settype(&attr,PTHREAD_MUTEX_RECURSIVE_NP);

pthread_mutex_init(&mutex,&attr);

pthread_mutex_destroy(&attr);

 

前面我们提到在调用pthread_mutex_lock()的时候,如果此时mutex已经被其他线程上锁,那么该操作将会一直阻塞在这个地方。如果我们此时不想一直阻塞在这个地方,那么可以调用下面函数:

pthread_mutex_trylock()

如果此时互斥量没有被上锁,那么pthread_mutex_trylock()将会返回0,并会对该互斥量上锁。如果互斥量已经被上锁,那么会立刻返回EBUSY。

 

上面谈到的是使用互斥量。如果碰到下面这种情况,该怎么办了?

还是上面程序中提到的工作链表。此时必然有一个生产者线程,用于往链表里添加节点。如果这一段时间没有工作,那么工作线程将会不停的调用lock,unlock操作。而这样的操作毫无疑义。

在这里系统给我们提供了另外一种同步机制,信号灯,Semaphore。

信号灯其实就是一个计数器,也是一个整数。每一次调用wait操作将会使semaphore值减一,而如果semaphore值已经为0,则wait操作将会阻塞。每一次调用post操作将会使semaphore值加一。将这些操作用到上面的问题中。工作线程每一次调用wait操作,如果此时链表中没有节点,则工作线程将会阻塞,直到链表中有节点。生产者线程在每次往链表中添加节点后调用post操作,信号灯值会加一。这样阻塞的工作线程就会停止阻塞,继续往下执行。

信号灯的类型为sem_t。在声明后必须调用sem_init()。需要传递两个参数,第一个参数就是你之前声明的sem_t变量,第二个必须为0。当你不再需要信号灯时,你必须调用sem_destroy()来释放资源。

等待信号灯的操作为sem_wait()。投递一个信号的操作为sem_wait()。和互斥量一样,等待信号灯也有一个非阻塞的操作,sem_trywait()。该操作在没有信号灯的时候返回EAGAIN。

下面是一个结合了互斥量和信号灯的例子:

#include <malloc.h>

#include <pthread.h>

#include <semaphore.h>

struct job {

/* Link field for linked list. */

struct job* next;

/* Other fields describing work to be done... */

};

/* A linked list of pending jobs. */

struct job* job_queue;

/* A mutex protecting job_queue. */

pthread_mutex_t job_queue_mutex = PTHREAD_MUTEX_INITIALIZER;

/* A semaphore counting the number of jobs in the queue. */

sem_t job_queue_count;

/* Perform one-time initialization of the job queue. */

void initialize_job_queue ()

{

/* The queue is initially empty. */

job_queue = NULL;

/* Initialize the semaphore which counts jobs in the queue. Its

initial value should be zero. */

sem_init (&job_queue_count, 0, 0);

}

/* Process queued jobs until the queue is empty. */

void* thread_function (void* arg)

{

while (1) {

struct job* next_job;

/* Wait on the job queue semaphore. If its value is positive,

indicating that the queue is not empty, decrement the count by

1. If the queue is empty, block until a new job is enqueued. */

sem_wait (&job_queue_count);

/* Lock the mutex on the job queue. */

pthread_mutex_lock (&job_queue_mutex);

/* Because of the semaphore, we know the queue is not empty. Get

the next available job. */

next_job = job_queue;

/* Remove this job from the list. */

job_queue = job_queue->next;

/* Unlock the mutex on the job queue because we’re done with the

queue for now. */

pthread_mutex_unlock (&job_queue_mutex);

/* Carry out the work. */

process_job (next_job);

/* Clean up. */

free (next_job);

}

return NULL;

}

/* Add a new job to the front of the job queue. */

void enqueue_job (/* Pass job-specific data here... */)

{

struct job* new_job;

/* Allocate a new job object. */

new_job = (struct job*) malloc (sizeof (struct job));

/* Set the other fields of the job struct here... */

/* Lock the mutex on the job queue before accessing it. */

pthread_mutex_lock (&job_queue_mutex);

/* Place the new job at the head of the queue. */

new_job->next = job_queue;

job_queue = new_job;

/* Post to the semaphore to indicate that another job is available. If

threads are blocked, waiting on the semaphore, one will become

unblocked so it can process the job. */

sem_post (&job_queue_count);

/* Unlock the job queue mutex. */

pthread_mutex_unlock (&job_queue_mutex);

}

 

 

 

下面说一下第三种同步机制―条件变量。

如果现在在等待一个信号。如果该信号被设置,则继续运行。如果没有条件变量,我们将会不停的去查询该信号是否被设置,这样就会浪费大量的cpu。而通过使用条件变量,我们就可以将等待信号的线程阻塞,直到有信号的时候再去唤醒它。

条件变量的类型是pthread_cond_t。

下面简单说一下如何使用条件变量。

l         声明pthread_cond_t变量后,调用pthread_cond_init()函数,第一个参数为之前声明的变量。第二个参数在Linux中不起作用。

l         声明一个pthread_mutex_t变量,并调用pthread_mutex_init()初始化。

l         调用pthread_cond_signal(),发出信号。如果此时有线程在等待该信号,那么该线程将会唤醒。如果没有,该信号就会别忽略。

l         如果想唤醒所有等待该信号的线程,调用pthread_cond_broadcast()。

l         调用pthread_cond_wait()等待信号。如果没有信号,线程将会阻塞,直到有信号。该函数的第一个参数是条件变量,第二个参数是一个mutex。在调用该函数之前必须先获得互斥量。如果线程阻塞,互斥量将立刻会被释放。

下面给出一个简单的使用例子。

#include <pthread.h>

#include <stdio.h>

 

pthread_mutex_t mutex;

pthread_cond_t cond;

int flag;

void init()

{

pthread_mutex_init(&mutex,NULL);

pthread_cond_init(&cond,NULL);

flag=0;

}

 

void * Thread_Function(void * arg)

{

//loop infinitely

while(1)

{

      pthread_mutex_lock(&mutex);

      while(!flag)

           pthread_cond_wait(&cond,&mutex);

      pthread_mutex_unlock(&mutex);

 

      do_some_work();

}

}

 

void SetFlag()

{

      pthread_mutex_lock(&mutex);

      flag=1;

      pthread_cond_signal(&cond);

      pthread_mutex_unlock(&mutex);

}

 
关于线程同步的技术先说到这个地方。





操作系统实验用的程序,保留一个。没什么技术含量,请见谅^_^

这个程序就是解决那个著名的“生产者―消费者”的问题,貌似线程的同步都讲这个问题,Java里面也有,不过操作起来貌似要更简单一点。

在这个程序里subp1()用来生产一个 int 数据,subp2()用来获取这个整数。首先是subp1()生产一个数据,subp2()再去获取这个数据。subp2()获取数据的首要条件是 subp1()已经生产了一个新的数据,subp1()生产一个新数据的前提是subp2()已经获得了 subp1()生产的前一个数据。
引用内容:
/*thread synchronization*/
/*thread.c*/
/*Afdream.com*/
/*2005-12-20*/
#include <stdio.h>
#include <stdlib.h>
#include <pthread.h>
#include <sys/types.h>
#include <linux/sem.h>
#include <semaphore.h>

/*function of p opration*/
void P(int semid,int index)
{
    struct sembuf sem;
    sem.sem_num = index;
    sem.sem_op = -1;
    //mark of option:0 or IPC_NOWAIT and so on
    sem.sem_flg = 0;
    //the '1' in this sentence means the number of commands
    semop(semid,&sem,1);       
    return;
}

/*function of v opration*/
void V(int semid,int index)
{
    struct sembuf sem;
    sem.sem_num = index;
    sem.sem_op =  1;
    sem.sem_flg = 0;
    semop(semid,&sem,1);
    return;
}

int semid;
pthread_t p1,p2;
int sharedInt=0;
void *subp1();
void *subp2();

/*in Linux,the return of the function main must be int,cann't be void*/
int main()
{
    union semun  semopts;
    int res;
    /*请求两个信号量*/
    semid = semget(300,2,IPC_CREAT|0666);
    if (semid<0) {
        printf("error");
        return;
    }
    /*初始化第一个信号量的值为1*/
    semopts.val = 1;
    res=semctl(semid,0,SETVAL,semopts);
    /*初始化第二个信号量的值为0*/
    semopts.val = 0;
    res=semctl(semid,1,SETVAL,semopts);
    if (res < 0) return;
    /*创建两个线程*/
    pthread_create(&p2,NULL,subp2,NULL);
                pthread_create(&p1,NULL,subp1,NULL);
    /*等待两个线程结束*/
    pthread_join(p1,NULL);
    pthread_join(p2,NULL);
    semctl(semid,0,IPC_RMID,0);
}

/*produce number*/
void *subp1()
{
    int i,j;
    for (i=0;  i<10;i++) {
        sleep(i+1);
        printf("\nready to produce!\n");
        P(semid,0);
        sharedInt++;
        printf("have produced %d!\n",sharedInt);
        V(semid,1);
    }
    return;
}

/*get number*/
void *subp2()
{
    int i,j;
    for (i=0;i<10;i++) {
        sleep(10-i);
        printf("\nready to get!\n");
        P(semid,1);
        printf("have got %d!\n",sharedInt);
        V(semid,0);
    }
    return;
}










posted on 2007-02-08 12:12 Khan 阅读(2273) 评论(0)  编辑 收藏 引用 所属分类: GCC/G++跨平台开发

只有注册用户登录后才能发表评论。
网站导航: 博客园   IT新闻   BlogJava   知识库   博问   管理