ACE基本的多线程编程（Basic Multithreaded Programming）

ACE基本的多线程编程（Basic Multithreaded Programming）

标签: Programming ACE Basic 线程 2008-12-10 19:23

从ACE_Task_Base或ACE_Task派生你的线程对象，重载virtual int svc (void);虚函数，那就是线程的入口函数。
#include <ace/OS.h>
#include <ace/Task.h>

class HA_CommandHandler : public ACE_Task_Base
{
public:
virtual int svc(void)
{
ACE_DEBUG((LM_DEBUG,
ACE_TEXT("(%t) Handler Thread running/n")));
ACE_OS::sleep(4);
return 0;
}
};

int ACE_TMAIN(int, ACE_TCHAR *[])
{
ACE_DEBUG((LM_DEBUG,
ACE_TEXT("(%t) Main Thread running/n")));

HA_CommandHandler handler;
int result = handler.activate();
ACE_ASSERT(result == 0);

handler.wait();
return 0;
}

在激活线程(activate()方法)之后，主线程会调用处理器对象上的wait()方法，等待其线程完成，然后在继续执行，并退出main()函数。

使用线程互斥体ACE_Thread_Mutex
如果成功获得互斥体，进行获取的线程（acquire()）继续向前执行；否则它就会阻塞，直到该互斥体的持有者释放它（release()）为止。
#include <ace/OS.h>
#include <ace/Log_Msg.h>
#include <ace/Task.h>
#include <ace/Thread_Mutex.h>

class HA_Device_Repository
{
public:
HA_Device_Repository()
{ }

void update_device(int device_id)
{
//使用守卫
//ACE_GUARD(ACE_Thread_Mutex, mon, mutex_);
ACE_Guard<ACE_Thread_Mutex> guard(this->mutex_);

//mutex_.acquire();
ACE_DEBUG((LM_DEBUG,
ACE_TEXT("(%t) Updating device %d/n"),
device_id));
ACE_OS::sleep(1);
//mutex_.release();
}
private:
ACE_Thread_Mutex mutex_;
};
class HA_CommandHandler : public ACE_Task_Base
{
public:
enum {NUM_USES = 10};

HA_CommandHandler(HA_Device_Repository& rep) : rep_(rep)
{ }

virtual int svc(void)
{
ACE_DEBUG((LM_DEBUG,
ACE_TEXT("(%t) Handler Thread running/n")));
for (int i=0; i < NUM_USES; i++)
this->rep_.update_device(i);
return 0;
}

private:
HA_Device_Repository& rep_;
};

int ACE_TMAIN(int, ACE_TCHAR *[])
{
HA_Device_Repository rep;
HA_CommandHandler handler1(rep);
HA_CommandHandler handler2(rep);
handler1.activate();
handler2.activate();

handler1.wait();
handler2.wait();

return 0;
}

使用守卫（Using Guards）
当然你在上面的代码中你已经看到了我已经使用了守卫（Guards）。
在许多的情况下，异常情况会在本可以完好运行的代码中造成死锁（忽略了某个异常路经、忘记释放互斥体）。
守卫（Guards）基于一种常见的C++惯用手法：把构造器和析构器用于资源的获取和释放。
在栈上使用守卫（Guards），你就总能保证锁的释放，不管你的代码所走的是什么样的非正常路径。
ACE Guard Classes

Guard

Description

ACE_Guard<T>	Uses the acquire() and release() methods of lock class T during guard creation and destruction. Thus, you get the semantics of acquire() and release() methods for the specified type T.
ACE_Read_Guard<T>	Uses acquire_read() for acquisition instead of the regular acquire().
ACE_Write_Guard<T>	Uses acquire_write() for acquisition instead of the regular acquire().
ACE_TSS_Guard<T>	Allocates the guard on the heap and keeps a reference to it in thread-specific storage. This ensures that the lock is always released even if the thread exits explicitly, using ACE_Thread::exit().
ACE_TSS_Read_Guard<T>	Read version of a thread-specific guard.
ACE_TSS_Write_Guard<T>	Write version of a thread-specific guard.

当然这里有相应的宏

The following guard macros do not return values:

ACE_GUARD (LockType, GuardName, LockObject)

ACE_WRITE_GUARD (LockType, GuardName, LockObject)

ACE_READ_GUARD (LockType, GuardName, LockObject)

These guard macros return ReturnValue on an error:

ACE_GUARD_RETURN (LockType, GuardName, LockObject, ReturnValue)

ACE_WRITE_GUARD_RETURN (LockType, GuardName, LockObject, ReturnValue)

ACE_READ_GUARD_RETURN (LockType, GuardName, LockObject, ReturnValue)

LockType用作守卫类模板中的T，GuardName是所创建的守卫对象名称，LockObject是守卫所引用的锁对象，而ReturnValue是出错时的返回值（so你的函数需要有返回值）。

任务间的通信
两个宽泛的范畴：
1）状态变化或事件通知；
使用条件变量ACE_Condition模板类。
获取互斥体，检查系统是否处在所需状态中，所需条件为真----如果是这样，就执行所需操作，然后释放互斥体，完成后更新后，调用条件变量signal()方法；所需条件为假----调用条件变量的wait()方法，等待系统状态发生变化。
#include <ace/OS.h>
#include <ace/Log_Msg.h>
#include <ace/Thread_Mutex.h>
#include <ace/Condition_T.h>
#include <ace/Task.h>

class HA_Device_Repository
{
private:
ACE_Task_Base* owner_;
//ACE_Thread_Mutex mutex_;
public:
HA_Device_Repository() : owner_(0)
{ }

void update_device(int device_id)
{
////使用守卫
//ACE_Guard<ACE_Thread_Mutex> guard(this->mutex_);

ACE_DEBUG((LM_DEBUG,
ACE_TEXT("(%t) Updating device %d/n"),
device_id));
ACE_OS::sleep(1);
}

int is_free(void)
{
return (this->owner_ == 0);
}

int is_owner (ACE_Task_Base* tb)
{
return (this->owner_ == tb);
}

ACE_Task_Base* get_owner(void)
{
return this->owner_;
}

void set_owner(ACE_Task_Base* owner)
{
this->owner_ = owner;
}
};
class HA_CommandHandler : public ACE_Task_Base
{
private:
HA_Device_Repository& rep_;
ACE_Thread_Mutex& mutex_;
ACE_Condition<ACE_Thread_Mutex>& waitCond_;

public:
enum {NUM_USES = 10};

HA_CommandHandler(HA_Device_Repository& rep,
ACE_Condition<ACE_Thread_Mutex>& wait,
ACE_Thread_Mutex& rep_mutex)
: rep_(rep),
waitCond_(wait),
mutex_(rep_mutex)
{ }

virtual int svc(void)
{
ACE_DEBUG((LM_DEBUG,
ACE_TEXT("(%t) Handler Thread running/n")));
for (int i=0; i < NUM_USES; i++)
{
this->mutex_.acquire();
while (!this->rep_.is_free())
//阻塞，进入休眠
this->waitCond_.wait();
this->rep_.set_owner(this);
this->mutex_.release();

this->rep_.update_device(i);

ACE_ASSERT(this->rep_.is_owner(this));
this->rep_.set_owner(0);

//让阻塞的进程苏醒过来
this->waitCond_.signal();

//让苏醒过来的进程有机会获得条件变量
ACE_OS::sleep(1);

}
return 0;
}
};
int ACE_TMAIN(int, ACE_TCHAR *[])
{
HA_Device_Repository rep;
ACE_Thread_Mutex rep_mutex;
ACE_Condition<ACE_Thread_Mutex> wait(rep_mutex);

HA_CommandHandler handler1(rep, wait, rep_mutex);
HA_CommandHandler handler2(rep, wait, rep_mutex);

handler1.activate();
handler2.activate();

handler1.wait();
handler2.wait();

return 0;
}
2）消息（数据）传递（Message Passing）。



消息块（Message Blocks）
ACE_Message_Block 一种高效的数据容器，可以用来高效的存储和共享消息。支持引用计数和数据共享特性。
rd_ptr()指针，指向要读取的下一个字节，；
wr_prt()指针，指向下一个可用的空字节；
copy()方法把数据复制到消息块；
msg_type()修改类型字段。
一旦你使用完了消息块，要用release()方法释放它，使引用计数减一，当引用计数到达0时，ACE会自动释放这个块分配的内存。

使用消息队列ACE_Message_Block
ACE_Task模板类含有一个ACE_Message_Block，你可以传入模板参数，ACE_MT_USE，让ACE_Message_Block是多线程安全的FIFO。
ACE_Task::putq()消息块入队；
ACE_Task::getq()消息块出队。

这里有多种ACE_Message_Block子类，有不同的特征：

Name

Description

ACE_Dynamic_Message_Queue	A priority-queue implementation that dynamically readjusts the priority of a message, using variants of the earliest-deadline-first scheme and a laxity (time to deadline minus worst-case execution time) schemes. For details, see the ACE reference documentation.
ACE_Message_Queue_Vx	Wrapper around the Wind River VxWorks message queue facility.
ACE_Message_Queue_Ex	An even more type-safe version of ACE_Message_Queue.
ACE_Message_Queue_NT	Implementation that is built on Windows NT's I/O completion port features.[a]