ACE程序员教程-第4章 线程管理：ACE的同步和线程管理机制

　　栅栏有一个好名字，因为它恰切地描述了栅栏应做的事情。一组线程可以使用栅栏来进行共同的相互同步。组中的每个线程各自执行，直到到达栅栏，就阻塞在那里。在所有相关线程到达栅栏后，它们就全部继续它们的执行。就是说，它们一个接一个地阻塞，等待其他的线程到达栅栏；一旦所有线程都到达了它们的执行路径中的“栅栏点”，它们就一起重新启动。

#include "ace/Thread.h"

#include "ace/Synch.h"

static int number=0;

static int seed=0;

class Args

{

public:

Args(ACE_Barrier *barrier): barrier_(barrier){}

ACE_Barrier *barrier_;

};

static void*

worker(void *arguments)

{

Args *arg= (Args*)arguments;

ACE_DEBUG((LM_DEBUG,"Thread (%t) Created to do some work\n"));

::number++;

//Work

ACE_OS::sleep(ACE_OS::rand()%2);

//Exiting now

ACE_DEBUG((LM_DEBUG,

"\tThread (%t) Done! \n\tThe number is now: %d\n",number));

　

//Let the barrier know we are done.

arg->barrier_->wait();

ACE_DEBUG((LM_DEBUG,"Thread (%t) is exiting \n"));

return 0;

}

int main(int argc, char *argv[])

{

if(argc<2)

{

ACE_DEBUG((LM_DEBUG,"Usage: %s \n", argv[0]));

ACE_OS::exit(1);

}

int n_threads=ACE_OS::atoi(argv[1]);

ACE_DEBUG((LM_DEBUG,"Preparing to spawn %d threads",n_threads));

//Setup the random number generator

ACE_OS::srand(::seed);

//Setup arguments for threads

ACE_Barrier barrier(n_threads);

Args arg(&barrier);

//Spawn off n_threads number of threads

for(int i=0; i

{

if(ACE_Thread::spawn((ACE_THR_FUNC)worker,

(void*)&arg,THR_DETACHED|THR_NEW_LWP)==-1)

ACE_DEBUG((LM_DEBUG,"Error in spawning thread\n"));

}

//Wait for all the other threads to let the main thread

// know that they are done using the barrier

barrier.wait();

ACE_DEBUG((LM_DEBUG,"(%t)Other threads are finished. Program exiting..\n"));

ACE_OS::sleep(2);

}

　　下面的例子演示此类的用法：

#include "ace/Synch.h"

//Global mutable and shared data on which we will perform simple

//arithmetic operations which will be protected.

ACE_Atomic_Op foo;

//The worker threads will start from here.

static void* worker(void *arg)

{

ACE_UNUSED_ARG(arg);

foo=5;

ACE_ASSERT (foo == 5);

++foo;

ACE_ASSERT (foo == 6);

--foo;

ACE_ASSERT (foo == 5);

foo += 10;

ACE_ASSERT (foo == 15);

foo -= 10;

ACE_ASSERT (foo == 5);

foo = 5L;

ACE_ASSERT (foo == 5);

return 0;

}

int main(int argc, char *argv[])

{

//spawn threads as in previous examples

}

　　你也许想知道为什么这样的算术原语（比如像上例中那样）需要同步。你一定认为增减运算本来就是原子的。

#include "ace/Thread_Manager.h"

#include "ace/Get_Opt.h"

static void* taskone(void*)

{

ACE_DEBUG((LM_DEBUG,"Thread:(%t)started Task one! \n"));

ACE_OS::sleep(2);

ACE_DEBUG((LM_DEBUG,"Thread:(%t)finished Task one!\n"));

return 0;

}

static void* tasktwo(void*)

{

ACE_DEBUG((LM_DEBUG,"Thread:(%t)started Task two!\n"));

ACE_OS::sleep(1);

ACE_DEBUG((LM_DEBUG,"Thread:(%t)finished Task two!\n"));

return 0;

}

static void print_usage_and_die()

{

ACE_DEBUG((LM_DEBUG,"Usage program_name

-a -b"));

ACE_OS::exit(1);

}

int main(int argc, char* argv[])

{

int num_task_1;

int num_task_2;

if(argc<3)

print_usage_and_die();

　

ACE_Get_Opt get_opt(argc,argv,"a:b:");

char c;

while( (c=get_opt())!=EOF)

{

switch(c)

{

case ’a’:

num_task_1=ACE_OS::atoi(get_opt.optarg);

break;

case ’b’:

num_task_2=ACE_OS::atoi(get_opt.optarg);

break;

default:

ACE_ERROR((LM_ERROR,"Unknown option\n"));

ACE_OS::exit(1);

}

}

//Spawn the first set of threads that work on task 1.

if(ACE_Thread_Manager::instance()->spawn_n(num_task_1,

(ACE_THR_FUNC)taskone,//Execute task one

0, //No arguments

THR_NEW_LWP, //New Light Weight Process

ACE_DEFAULT_THREAD_PRIORITY,

1)==-1) //Group ID is 1

ACE_ERROR((LM_ERROR,

"Failure to spawn first group of threads: %p \n"));

　

//Spawn second set of threads that work on task 2.

if(ACE_Thread_Manager::instance()->spawn_n(num_task_2,

(ACE_THR_FUNC)tasktwo,//Execute task one

0, //No arguments

THR_NEW_LWP, //New Light Weight Process

ACE_DEFAULT_THREAD_PRIORITY,

2)==-1)//Group ID is 2

ACE_ERROR((LM_ERROR,

"Failure to spawn second group of threads: %p \n"));

　

//Wait for all tasks in grp 1 to exit

ACE_Thread_Manager::instance()->wait_grp(1);

ACE_DEBUG((LM_DEBUG,"Tasks in group 1 have exited! Continuing \n"));

//Wait for all tasks in grp 2 to exit

ACE_Thread_Manager::instance()->wait_grp(2);

ACE_DEBUG((LM_DEBUG,"Tasks in group 2 have exited! Continuing \n"));

}

// Test out the group management mechanisms provided by the

// ACE_Thread_Manager, including the group suspension and resumption,

//and cooperative thread cancellation mechanisms.

#include "ace/Thread_Manager.h"

static const int DEFAULT_THREADS = ACE_DEFAULT_THREADS;

static const int DEFAULT_ITERATIONS = 100000;

static void *

worker (int iterations)

{

for (int i = 0; i < iterations; i++)

{

if ((i % 1000) == 0)

{

ACE_DEBUG ((LM_DEBUG,

"(%t) checking cancellation before iteration %d!\n", i));

　

//Before doing work check if you have been canceled. If so don’t

//do any more work.

if (ACE_Thread_Manager::instance ()->testcancel

(ACE_Thread::self ()) != 0)

{

ACE_DEBUG ((LM_DEBUG,

"(%t) has been canceled before iteration %d!\n",i));

break;

}

}

}

return 0;

}

int main (int argc, char *argv[])

{

int n_threads = argc > 1 ? ACE_OS::atoi (argv[1]) : DEFAULT_THREADS;

int n_iterations = argc > 2 ? ACE_OS::atoi (argv[2]) : DEFAULT_ITERATIONS;

ACE_Thread_Manager *thr_mgr = ACE_Thread_Manager::instance ();

//Create a group of threads n_threads that will execute the worker

//function the spawn_n method returns the group ID for the group of

//threads that are spawned. The argument n_iterations is passed back

//to the worker. Notice that all threads are created detached.

int grp_id = thr_mgr->spawn_n (n_threads, ACE_THR_FUNC (worker),

(void *) n_iterations,

THR_NEW_LWP | THR_DETACHED);

　

// Wait for 1 second and then suspend every thread in the group.

ACE_OS::sleep (1);

ACE_DEBUG ((LM_DEBUG, "(%t) suspending group\n"));

if (thr_mgr->suspend_grp (grp_id) == -1)

ACE_ERROR ((LM_DEBUG, "(%t) %p\n", "Could not suspend_grp"));

　

// Wait for 1 more second and then resume every thread in the

// group.

ACE_OS::sleep (1);

ACE_DEBUG ((LM_DEBUG, "(%t) resuming group\n"));

if (thr_mgr->resume_grp (grp_id) == -1)

ACE_ERROR ((LM_DEBUG, "(%t) %p\n", "resume_grp"));

　

// Wait for 1 more second and then cancel all the threads.

ACE_OS::sleep (ACE_Time_Value (1));

ACE_DEBUG ((LM_DEBUG, "(%t) canceling group\n"));

if (thr_mgr->cancel_grp (grp_id) == -1)

ACE_ERROR ((LM_DEBUG, "(%t) %p\n", "cancel_grp"));

　

// Perform a barrier wait until all the threads have shut down.

thr_mgr->wait ();

return 0;

}

#include "ace/Synch.h"

#include "ace/Thread_Manager.h"

class DataType

{

public:

DataType():data(0){}

void increment(){ data++;}

void set(int new_data){ data=new_data;}

void decrement(){ data--;}

int get(){return data;}

private:

int data;

};

ACE_TSS data;

static void* thread1(void*)

{

data->set(10);

ACE_DEBUG((LM_DEBUG,"(%t)The value of data is %d \n",data->get()));

for(int i=0;i<5;i++)

data->increment();

ACE_DEBUG((LM_DEBUG,"(%t)The value of data is %d \n",data->get()));

return 0;

}

static void * thread2(void*)

{

data->set(100);

ACE_DEBUG((LM_DEBUG,"(%t)The value of data is %d \n",data->get()));

for(int i=0; i<5;i++)

data->increment();

ACE_DEBUG((LM_DEBUG,"(%t)The value of data is %d \n",data->get()));

return 0;

}

int main(int argc, char*argv[])

{

//Spawn off the first thread

ACE_Thread_Manager::instance()->spawn((ACE_THR_FUNC)thread1,0,THR_NEW_LWP|

THR_DETACHED);

　

//Spawn off the second thread

ACE_Thread_Manager::instance()->spawn((ACE_THR_FUNC)thread2,0,THR_NEW_LWP| THR_DETACHED);

　

//Wait for all threads in the manager to complete.

ACE_Thread_Manager::instance()->wait();

ACE_DEBUG((LM_DEBUG,"Both threads done.Exiting.. \n"));

}