桃源谷

心灵的旅行

人生就是一场旅行,不在乎旅行的目的地,在乎的是沿途的风景和看风景的心情 !
posts - 32, comments - 42, trackbacks - 0, articles - 0
  C++博客 :: 首页 :: 新随笔 :: 联系 :: 聚合  :: 管理

Linux中处理来自共享对象的同步事件

怎么利用设计模式来更有效的使用共享内存

 

级别:中等

 Sachin Agrawal (sachin_agrawal@in.ibm.com), Senior Software Engineer, IBM Software Labs, India
Swati P. Udas (swatudas@in.ibm.com), Software Engineer, IBM

 

10 Nov 2005

 在高级语言例如C++中有效的使用共享内存并不是一件浅显易懂的事情,但是它也能克服这些内在的困难。这篇文章描述了在Linux上使用共享内存的两个C++设计模式并包含了样例代码,以及给读者打开了更有效的进程间通信的一扇门。

 在面向对象系统中,当一个对象接收到一个消息时它能够发送一套事件。这些事件主要在同步模式下被处理。这个调用进程或者线程在发送消息调用完成之前,发送给对象一个消息和处理事件。然而,如果这个对象送出这些被更多的进程共享以及驻留在内存里的事件,情况就稍微的发生了一些变化。

 这篇文章用两个C++的设计模式详细的描述了上述的情况,并且用一些例子程序阐明了解决方案。  

  1. 我们首先描述了没有使用共享内存的例子程序。
  2. 其次作了一些改动去使用共享内存,这里使用的是第一种设计模式
  3. 最后,阐述了怎么完成进程间通信,使用的是第二中设计模式  

你能应用这些设计模式中的全部概念悼任何的机器构架,操作系统和编译器上。我们使用的是32Intel®构架的RedHat Linux 7.1发行版 ,使用GNU C++编译器的版本是3.2.3

 

没有共享内存 

让我们开始一个没有使用共享内存的例子程序:

 1Listing 1. common.h
 2
 3#ifndef __COMMON_H__
 4#define __COMMON_H__
 5 
 6
 7class IObjectWithEvents
 8{
 9public:
10   class IEventSink
11   {
12   public:
13      virtual void OnEvent(pid_t pid, const char * msg) = 0;
14   }

15
16   static IObjectWithEvents * getInstance();
17   virtual bool AddEventHandler(IEventSink * pEI) = 0;
18   virtual void SendMessage() = 0;
19
20}

21
22#endif //__COMMON_H__

 

IObjectWithEvents类接口中包含了定义了OnEvent()方法的IEventSink这个嵌入类,这是一个接受发送者pid和字符串消息的事件处理器。getInstance()方法返回一个共享内存中的对象的引用,AddEventHandler()是注册一个事件处理器,SendMessage()发送一个消息到对象上,没有任何共享内存的引用,下面的Listing 2中列出了IobjectWithEvents的程序代码:

  

 1 Listing 2. shm-client1.cpp
 2 
 3 #include <iostream>
 4 #include <sys/types.h>
 5 #include <unistd.h>
 6 #include "common.h" 
 7 
 8 #define HERE __FILE__ << ":" << __LINE__ << " "
 9 
10 using namespace std; 
11 
12 class EventSink : public IObjectWithEvents::IEventSink
13 {
14 public:
15    void OnEvent(pid_t pid, const char * msg)
16    {
17       cout << HERE << "Message from pid(" << pid << ")\t : " << msg << endl;
18    }
19 };
20 
21 
22 int main()
23 {
24    IObjectWithEvents * powe = IObjectWithEvents::getInstance(); 
25 
26    EventSink sink;
27    powe->AddEventHandler(&sink); 
28 
29    powe->SendMessage();
30    return 0;
31 

 EventSink类提供了事件处理器的实现,在主函数中显示了发送消息和处理事件的标准顺序。

Listing3中列出了ObjectWithEvents的典型实现代码:

  1 Listing 3. ObjectWithEvents.h 
  2 
  3 #include "common.h" 
  4 
  5 class ObjectWithEvents : public IObjectWithEvents
  6 {
  7 public:
  8    // We assume singleton design pattern for illustration
  9    static ObjectWithEvents * ms_pObjectWithEvents; 
 10 
 11    ObjectWithEvents(); 
 12 
 13    //the implementation for IObjectWithEvents
 14    void FireEvent();
 15    virtual bool AddEventHandler(IEventSink * pEI);
 16    virtual void SendMessage(); 
 17 
 18    //Collection for maintaining events
 19    enum { MAX_EVENT_HANDLERS = 16, };
 20    long m_npEI;
 21    IEventSink * m_apEI[MAX_EVENT_HANDLERS];
 22    pid_t m_alPID[MAX_EVENT_HANDLERS];
 23 }; 
 24 
 25 
 26 Listing 4. ObjectWithEvents.cpp 
 27 
 28 #include <iostream>
 29 #include <sys/types.h>
 30 #include <sys/shm.h>
 31 #include <unistd.h>
 32 #include <pthread.h>
 33 #include "ObjectWithEvents.h"
 34 
 35 using namespace std;
 36 
 37 ObjectWithEvents * ObjectWithEvents::ms_pObjectWithEvents = NULL;
 38 
 39 IObjectWithEvents * IObjectWithEvents::getInstance()
 40 {
 41    // the following commented code is for illustration only.
 42 
 43    /*
 44    if (NULL == ObjectWithEvents::ms_pObjectWithEvents)
 45    {
 46       ObjectWithEvents::ms_pObjectWithEvents = new ObjectWithEvents();
 47    }
 48    */ 
 49 
 50    return ObjectWithEvents::ms_pObjectWithEvents;
 51 
 52 
 53 ObjectWithEvents::ObjectWithEvents() : m_npEI(0)
 54 {
 55 
 56 }
 57  
 58 
 59 void ObjectWithEvents::FireEvent()
 60 {
 61    // iterate through the collection
 62    for (long i = 0; i < m_npEI; i++)
 63    {
 64       //Recheck for NULL
 65       if (0 != m_apEI[i])
 66       {
 67          // Fire the event
 68          m_apEI[i]->OnEvent(m_alPID[i], "");
 69       }
 70    } 
 71 
 72    return;
 73 
 74 
 75 bool ObjectWithEvents::AddEventHandler(IEventSink * pEI)
 76 {
 77    // NULL check
 78    if (NULL == pEI)
 79    {
 80       return false;
 81    }
 82 
 83    // check if there is space for this event handler
 84    if (MAX_EVENT_HANDLERS == m_npEI)
 85    {
 86       return false;
 87    } 
 88 
 89    // Add this event handler to the collection
 90    m_alPID[m_npEI] = getpid();
 91    m_apEI[m_npEI++= pEI; 
 92 
 93    return true;
 94 
 95 
 96 void ObjectWithEvents::SendMessage()
 97 {
 98    //Some processing
 99    //And then fire the event 
100 
101    FireEvent(); 
102 
103    return;
104 



你能使用下面的命令行来编译这些例子程序:

g++ -g -o shm_client shm_client1.cpp ObjectWithEvents.cpp

 当你运行shm_client时,将得到下面的输出:

$ ./shm_client shm_client1.cpp:16 Message from pid(3920) :

 

使用共享内存:没有事件缓存

 现在,对于在共享内存中实例ObjectWithEvents的实现作了以下的修改。

 1 Listing 5. Changes to ObjectWithEvents.cpp 
 2 
 3 
 4 // To add a declaration for the "new" operator:
 5 class ObjectWithEvents : public IObjectWithEvents{
 6 public:   void * operator new(unsigned int);
 7 };
 8 
 9 // To include an additional header for the Initializer class:
10 
11 #include "Initializer.h"
12   
13 
14 // To overload the operator "new":
15 
16 void * ObjectWithEvents::operator new(unsigned int)
17 {
18    return ms_pObjectWithEvents;
19 }
20  
21 
22 // Then, FireEvent is completely changed:
23 
24 
25 void ObjectWithEvents::FireEvent()
26 {
27    // We need to serialize all access to the collection by more than one process
28    int iRetVal = Initializer::LockMutex(); 
29 
30    if (0 != iRetVal)
31    {
32       return;
33    } 
34 
35    pid_t pid = getpid(); 
36 
37    // iterate through the collection and fire only events belonging to the current process
38    for (long i = 0; i < m_npEI; i++)
39    {
40       // Check whether the handler belongs to the current process.
41       if (pid != m_alPID[i])
42       {
43          continue;
44       } 
45 
46       //Recheck for NULL
47       if (0 != m_apEI[i])
48       {
49          m_apEI[i]->OnEvent(pid, "");
50       }
51    } 
52 
53    // release the mutex
54    if ((0 == iRetVal) && (0 != Initializer::UnlockMutex()))
55    {
56       // Deal with error.
57    } 
58 
59    return;
60 }
61  
62 
63 // The following are changes to ObjectWithEvents::AddEventHandler(): 
64 
65 // 1. Before accessing the collection, we lock the mutex: 
66 
67 int bRetVal = Initializer::LockMutex();
68 
69 if (0 != bRetVal)
70 {
71    return false;
72 }
73  
74 
75 // 2. After accessing the collection, we release the mutex: 
76 
77 if ((0 == bRetVal) && (0 != Initializer::UnlockMutex()))
78 {
79    // Deal with error.
80 

 

在共享内存中的示例化对象,定义了一个叫做Initializer的额外的类  

 

 1 Listing 6. Initializer.h 
 2 
 3 #ifndef __Initializer_H__
 4 #define __Initializer_H__ 
 5 
 6 class Initializer
 7 {
 8 public :
 9     int m_shmid;
10     static Initializer ms_Initializer;
11     Initializer(); 
12 
13     static pthread_mutex_t ms_mutex;
14     static int LockMutex();
15     static int UnlockMutex();
16 }; 
17 
18 #endif // __Initializer_H__ 

 

Initializer定义了共享内存id的变量m_shmid和对于同步事件处理器的一个信号量变量ms_mutex.

LockMutex()锁定互斥体,UnlockMutex()则解锁互斥体。

Listing7列出了Initializer的实现代码:

 

 1 Listing 7. Initializer.cpp 
 2 
 3 #include <iostream>
 4 #include <sys/types.h>
 5 #include <sys/shm.h>
 6 #include <unistd.h>
 7 #include <pthread.h>
 8 #include "Initializer.h"
 9 #include "ObjectWithEvents.h" 
10 
11 using namespace std;
12 
13 Initializer Initializer::ms_Initializer; 
14 
15 pthread_mutex_t Initializer::ms_mutex = PTHREAD_ERRORCHECK_MUTEX_INITIALIZER_NP; 
16 
17 Initializer::Initializer() : m_shmid(-1)
18 {
19    bool bCreated = false;
20    key_t key = 0x1234
21 
22    m_shmid = shmget(key,sizeof(ObjectWithEvents), 0666); 
23 
24    if (-1 == m_shmid)
25    {
26       if(ENOENT != errno)
27       {
28          cerr<<"Critical Error"<<endl;
29          return;
30       } 
31 
32       m_shmid = shmget(key, sizeof(ObjectWithEvents), IPC_CREAT|0666); 
33 
34       if (-1 == m_shmid )
35       {
36          cout << " Critical Error " << errno<< endl;
37          return;
38       }
39 
40       bCreated = true;
41    } 
42 
43    ObjectWithEvents::ms_pObjectWithEvents = (ObjectWithEvents*)shmat(m_shmid,NULL,0); 
44 
45    if (NULL == ObjectWithEvents::ms_pObjectWithEvents)
46    {
47       cout << " Critical Error " << errno << endl;
48       return;
49    }
50 
51    if (true == bCreated)
52    {
53       ObjectWithEvents * p = new ObjectWithEvents();
54    }
55 
56    // Create a mutex with no initial owner. 
57 
58    pthread_mutex_init(&ms_mutex, NULL);
59 }
60 
61  
62 
63 int Initializer::LockMutex()
64 {
65    // Request ownership of mutex.
66 
67    pthread_mutex_lock(&ms_mutex); 
68 
69    if(EDEADLK == errno)
70    {
71       cout << "DeadLock" << endl;
72       return -1;
73    }
74 
75    return 0;
76 
77 
78 int Initializer::UnlockMutex()
79 {
80    return pthread_mutex_unlock(&ms_mutex);
81 

 

 

如果共享内存不存在的话则创建它,并在共享内存里做成共享对象。如果共享内存已经存在的话,则略过构造共享对象。Initializer::m_shmid纪录标识符ObjectWithEvents::ms_pObjectWithEvents并记录共享对象的引用。  

即使在所有的进程从这个共享内存分离(detach)也不释放共享内存。让你用ipcrm命令显示的销毁它并能用ipcs命令快速察看共享内存的信息。用下面的命令编译成可执行程序:
g++ -g -o shm_client shm_client1.cpp ObjectWithEvents.cpp Initializer.cpp 

控制台上会有下面那样的输出信息:

Listing 8. The console dump

$ ./shm_client

shm_client1.cpp:16 Message from pid(4332)        :

$ ipcs
------ Shared Memory Segments --------
key        shmid      owner      perms      bytes      nattch     status
0x00001234 327686     sachin    666        136        0
$ ./shm_client

shm_client1.cpp:16 Message from pid(4333)        :

$ ipcrm -m 327686 

 

ObjectWithEvents实例中有一个能从很多进程中收集事件的收集器。它仅仅能发送当前进程注册的事件。设计模式中表明了以下的两点:

  • 任何访问事件收集器都要被互斥体对象保护  
  • 在发送之前事件都要通过进程ID进行过滤。  

 IPC的共享内存和事件缓存 

现在,让我们看看进程间通信的共享内存和事件缓存。如果事件被缓存到共享对象里,则他们在稍后会被过滤。接收的进程将根据事件查询共享对象。然后,通过一个同步模型,进程间的通信能被接受到,这是开发下面的设计模式的主要动机。

 

IobjectWithEvents中,像下面那样增加一对儿方法: 

 1 Listing 9. Adding methods to IobjectWithEvents 
 2 
 3 class IObjectWithEvents
 4 {
 5 public:
 6 
 7    virtual bool EnqueueEvent(const char * msg) = 0;
 8    virtual bool PollForEvents() = 0;
 9 }; 
10 

EnqueueEvent() 简单的增加到共享对象中的事件缓存中,并且PollForEvents()将接收这个缓存。

shm_client1将像下面那样被使用EnqueueEvent()

powe->EnqueueEvent("Message from shm_client1");  

 

shm_client2(实质上是shm_client1的拷贝)将像下面那样使用PollForEvents()

powe->EnqueueEvent("Message from shm_client2"); powe->PollForEvents();  

 

ObjectWithEvents的实现代码是下面那样:

Listing 10. Additions to ObjectWithEvents

 1 class ObjectWithEvents : public IObjectWithEvents
 2 {
 3 public:
 4    virtual bool EnqueueEvent(const char * msg);
 5    virtual bool PollForEvents(); 
 6 
 7    //The event cache
 8    enum { MAX_EVENTS = 16, MAX_EVENT_MSG = 256, };
 9    long m_nEvents;
10    pid_t m_alPIDEvents[MAX_EVENTS];
11    char m_aaMsgs[MAX_EVENTS][MAX_EVENT_MSG];
12 }; 
13 
14 
15 

 

新的构造函数变成:

ObjectWithEvents::ObjectWithEvents() : m_npEI(0), m_nEvents(0) { }

 

EnqueueEvent() 存储事件(例如,每一个被发送的事件的消息和进程号)到一个队列中,PollForEvents()则迭代整个队列并为队列中的事件一个一个的调用OnEvent().

Listing 11. EnqueueEvent

 

 

 1 bool ObjectWithEvents::EnqueueEvent(const char * msg)
 2 {
 3    if (NULL == msg)
 4    {
 5       return false;
 6    }
 7 
 8    if (MAX_EVENTS == m_nEvents)
 9    {
10       //IEventSink collection full
11       return false;
12    }
13 
14    int bRetVal = Initializer::LockMutex();
15 
16    if (0 != bRetVal)
17    {
18       return false;
19    }
20 
21    m_alPIDEvents[m_nEvents] = getpid();
22    strncpy(m_aaMsgs[m_nEvents++], msg, MAX_EVENT_MSG - 1);
23 
24    if ((0 == bRetVal) && (0 != Initializer::UnlockMutex()))
25    {
26       // Deal with error.
27    }
28 
29    return true;
30 }
31 
32 
33 bool ObjectWithEvents::PollForEvents()
34 {
35    if (0 == m_nEvents)
36    {
37       return true;
38    }
39 
40    int bRetVal = Initializer::LockMutex();
41 
42    if (0 != bRetVal)
43    {
44       return false;
45    }
46 
47    pid_t pid = getpid();
48 
49    for (long i = 0; i < m_npEI; i++)
50    {
51       // Does the handler belongs to current process ? 
52 
53       if (pid != m_alPID[i])
54       {
55          continue;
56       }
57 
58       //Recheck for NULL
59 
60       if (0 == m_apEI[i])
61       {
62          continue;
63       }
64 
65       for (long j = 0; j < m_nEvents; j++)
66       {
67          m_apEI[i]->OnEvent(m_alPIDEvents[j], m_aaMsgs[j]);
68       }
69    }
70 
71    if ((0 == bRetVal) && (0 != Initializer::UnlockMutex()))
72    {
73       // Deal with error.
74    } 
75 
76    return true;
77 

   

现在使用下面的命令变成新命令:
g++ -g -o shm_client1 shm_client1.cpp ObjectWithEvents.cpp Initializer.cpp
g++ -g -o shm_client2 shm_client2.cpp ObjectWithEvents.cpp Initializer.cpp

 

在你的控制台上将像下面那样输出消息:

 

Listing 12. Output from shm_client1 and shm_client2  

$ ./shm_client1

$ ./ipcs

------ Shared Memory Segments --------
key        shmid      owner      perms      bytes      nattch     status
0x00001234 360454     sachin    666        4300       0

$ ./shm_client2

shm_client2.cpp:16 Message from pid(4454)        : Message from shm_client1

shm_client2.cpp:16 Message from pid(4456)        : Message from shm_client2

下载 

 

 

Description 

Name

 

 

Size

 

 

Download method

 

 

Shared memory sample code 

sync_code.zip

 

 

4KB

 

 

FTP

 


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


我的个人简历第一页 我的个人简历第二页