NDK下IPC问题

由于AllJoyn的join session timeout问题一直无法解决，我们怀疑AllJoyn有些内部变量没有清理干净，因此考虑将AllJoyn相关功能放到一个单独的进程中，一旦join session timeout，就重启该进程。

这就涉及到IPC问题。

因为我们写的是通用模块，需要在DTV（arm平台linux）和Android(NDK)上运行，因此就必须考虑各种IPC手段的可用性了。

一开始我考虑直接使用socket进行IPC（肯定可用），但有人说消息队列、共享内存、管道之类的也可以用。于是俺就mmap+信号量实现了一个简单的IPC类：

/* 
 * File:   ipc.h
 * Author: raozf
 *
 * Created on 2013年9月17日, 上午11:04
 */

#ifndef IPC_H
#define    IPC_H

#include <semaphore.h>
#include <sys/mman.h>
#include <sys/stat.h>
#include <fcntl.h>

//semaphore name
const std::string IPC_SEM = "alljoyn_sem";
//mmap length
const size_t IPC_MMAP_LEN = 1 * 1024 * 1024;

/*
 * helper class for IPC  between ****forked**** processes.
 * implemented by semaphore and mmap
 */
class IPC
{
public:
    IPC():_mutex(NULL), _memory(NULL)
    {
        //unnamed semaphore should use sem_init()
        
        //open named semaphore, shared between processes
        _mutex = sem_open(IPC_SEM.c_str(), O_CREAT, O_RDWR, 1);
        if(_mutex == SEM_FAILED || _mutex == NULL)
        {
            LOGV("[ContextSharing]IPC::IPC() sem_open() failed. semaphore name:%s, %s", IPC_SEM.c_str(), strerror(errno));
            return;
        }
        
        /*
         * http://www.ibm.com/developerworks/cn/linux/l-ipc/part5/index1.html
         */
        //create a anonymous mmap
        _memory = (char*)mmap(NULL, IPC_MMAP_LEN, PROT_READ | PROT_WRITE, MAP_SHARED | MAP_ANONYMOUS, 0, 0);
        if(_memory == MAP_FAILED || _memory == NULL)
        {
            LOGV("[ContextSharing]IPC::IPC() mmap() failed. %s", strerror(errno));
            deinit();
            return;
        }
    }

    ~IPC()
    {
        deinit();
    }
    
    //send msg to other process
    //header+message
    //   int         char*
    void send(const std::string& msg)
    {
        *((int*)_memory) = msg.length();
        memcpy(_memory + sizeof(int), msg.c_str(), msg.length());
        sem_post(_mutex);
    }
    
    //receive from other process
    std::string recv()
    {
        sem_wait(_mutex);//will block
        return std::string(_memory + sizeof(int),  (int)(_memory));
    }

private:
    void deinit()
    {
        if(_mutex != NULL)
        {
            /* destory an unnamed semaphore, initialized by sem_init()
            //sem_destroy(_mutex);
             */
            
            // closes  the  named semaphore, but still existed in system kernel
            sem_close(_mutex);
            _mutex = NULL;
            
            //remove semaphore from system kernel(if it's reference is 0--which decremented  by sem_close())
            //removes  the  named  semaphore  referred  to  by name.  The semaphore name is removed immediately.  
            //The semaphore is destroyed once all other processes that have the semaphore open close it.
            sem_unlink(IPC_SEM.c_str());
        }
        
        if(_memory != NULL)
        {
            munmap(_memory, IPC_MMAP_LEN);
            _memory = NULL;
        }        
    }
    
private:
    sem_t* _mutex;
    char* _memory;
};

#endif    /* IPC_H */

辅助类，封装一个线程专门等待信号量、接收数据：

/* 
 * File:   ThreadedIPC.h
 * Author: raozf
 *
 * Created on 2013年9月17日, 下午3:21
 */

#ifndef THREADEDIPC_H
#define    THREADEDIPC_H

#include "Common/ThreadManager.h"
#include "delegate.h"
#include "ipc.h"

static const OID OID_ThreadedIPC = { 0x8f52a31f, 0x51d5, 0x462b, { 0xa9, 0x8d, 0x40, 0x70, 0xa3, 0xf6, 0xb5, 0x7f } };
class ThreadedIPC
    :public CObjectRoot<CObjectMultiThreadModel>
    ,public IThreadClient
{
public:
    PSFRESULT FinalConstruct()
    {
        PSFRESULT res = PSF_E_FAIL;
        
        CHK_PSFRESULT(_listen_thread.Initialize());
        _listen_thread.AddTask(this, NULL);
        
        CLEANUP:
            return res;
    }
    
    void FinalRelease()
    {
        _listen_thread.Terminate();
    }
    
    void Send(const std::string& msg)
    {
        _sharer.send(msg);
    }
    
private:
    void OnExecute(void* pArg)
    {
        while(true)
        {
            _delegate_recv(_sharer.recv());
        }
    }
    
    void OnTerminate(void* pArg)
    {
    }

public:
    delegate::Delegate<void(std::string)> _delegate_recv;
    
private:
    CTaskWorker _listen_thread;
    IPC _sharer;

    BEGIN_OBJECT_INTERFACE_MAP()
        OBJECT_INTERFACE_ENTRY(OID_ThreadedIPC, this);
    END_OBJECT_INTERFACE_MAP()
};

#endif    /* THREADEDIPC_H */

这里面用到了我们自己写的delegate和thread类（先忽略之）。

但在android上运行时却报错：

IPC::IPC() sem_open() failed. semaphore name:alljoyn_sem, Function not implemented

看来信号量在NDK下是用不了的。

而且，不仅仅信号量，共享内存、消息队列在NDK下都不能用。

参见讨论：https://groups.google.com/forum/#!topic/android-ndk/FzJIsJIxCX4

http://stackoverflow.com/questions/18603267/cross-process-locking-with-android-ndk

（但管道可用？http://stackoverflow.com/questions/8471552/porting-c-code-which-uses-fork-to-android）

Android源代码中的文档说明：http://www.netmite.com/android/mydroid/1.6/bionic/libc/docs/SYSV-IPC.TXT

（该文件在Android4。3中似乎已经移出该文档）

Android does not support System V IPCs, i.e. the facilities provided by the
following standard Posix headers:

  <sys/sem.h>   /* SysV semaphores */
  <sys/shm.h>   /* SysV shared memory segments */
  <sys/msg.h>   /* SysV message queues */
  <sys/ipc.h>   /* General IPC definitions */

The reason for this is due to the fact that, by design, they lead to global
kernel resource leakage.

原因就是防止内核资源泄露。

更重要的问题在于：fork()也尽量不要用。

道理很简单：我们不应该控制android的底层，这些api会造成系统的不稳定。

https://groups.google.com/forum/#!msg/android-platform/80jr-_A-9bU/nkzslcgVrfYJ

“Bear in mind that the Dalvik VM doesn't like fork() much, and goes into
conniptions if you try to do *any* work between the fork() and the
exec(). ”

https://groups.google.com/forum/#!topic/android-ndk/FzJIsJIxCX4

悲催！

 ----------------------------

ps:学到了一个命令“ipcs”，查看系统中共享内存、信号量状态：

$ ipcs

------ Shared Memory Segments --------
key shmid owner perms bytes nattch status

------ Semaphore Arrays --------
key semid owner perms nsems

------ Message Queues --------
key msqid owner perms used-bytes messages