QNX相關(guān)歷史文章：

• QNX簡介
• QNX Neutrino微內(nèi)核
• QNX IPC機(jī)制
• QNX進(jìn)程管理器
• QNX資源管理器
• QNX字符I/O
• QNX之編寫資源管理器（一）
• QNX之編寫資源管理器（二）
• QNX之編寫資源管理器（三）

Handling Read and Write Messages

這篇文章主要描述讀寫消息的處理馒索。

1. Handling the _IO_READ message

io_read處理函數(shù)負(fù)責(zé)處理_IO_READ消息命雀，將收到的數(shù)據(jù)返回給客戶端借杰，比如當(dāng)客戶端調(diào)用read()/readdir()/fread()/fgetc()等接口時(shí)。
消息的定義如下：

struct _io_read {
    uint16_t            type;
    uint16_t            combine_len;
    int32_t             nbytes;
    uint32_t            xtype;
};

typedef union {
    struct _io_read     i;
    /* unsigned char    data[nbytes];    */
    /* nbytes is returned with MsgReply  */
} io_read_t;

• combine_len退腥，用于消息組合；
• nbytes再榄，用于客戶端期望讀取的字節(jié)數(shù)狡刘；
• xtype，這個(gè)字段下文會(huì)介紹到困鸥，主要用于擴(kuò)展嗅蔬；

下邊是一個(gè)完整的消息讀取示例代碼：

#include <errno.h>
#include <stdio.h>
#include <stddef.h>
#include <stdlib.h>
#include <unistd.h>
#include <sys/iofunc.h>
#include <sys/dispatch.h>

int io_read (resmgr_context_t *ctp, io_read_t *msg, RESMGR_OCB_T *ocb);

static char                     *buffer = "Hello world\n";

static resmgr_connect_funcs_t   connect_funcs;
static resmgr_io_funcs_t        io_funcs;
static iofunc_attr_t            attr;

main(int argc, char **argv)
{
    /* declare variables we'll be using */
    resmgr_attr_t        resmgr_attr;
    dispatch_t           *dpp;
    dispatch_context_t   *ctp;
    int                  id;

    /* initialize dispatch interface */
    if((dpp = dispatch_create()) == NULL) {
        fprintf(stderr, "%s: Unable to allocate dispatch handle.\n",
                argv[0]);
        return EXIT_FAILURE;
    }

    /* initialize resource manager attributes */
    memset(&resmgr_attr, 0, sizeof resmgr_attr);
    resmgr_attr.nparts_max = 1;
    resmgr_attr.msg_max_size = 2048;

    /* initialize functions for handling messages */
    iofunc_func_init(_RESMGR_CONNECT_NFUNCS, &connect_funcs,
                     _RESMGR_IO_NFUNCS, &io_funcs);
    io_funcs.read = io_read;

    /* initialize attribute structure used by the device */
    iofunc_attr_init(&attr, S_IFNAM | 0666, 0, 0);
    attr.nbytes = strlen(buffer)+1;
    
    /* attach our device name */
    if((id = resmgr_attach(dpp, &resmgr_attr, "/dev/sample", _FTYPE_ANY, 0,
                 &connect_funcs, &io_funcs, &attr)) == -1) {
        fprintf(stderr, "%s: Unable to attach name.\n", argv[0]);
        return EXIT_FAILURE;
    }

    /* allocate a context structure */
    ctp = dispatch_context_alloc(dpp);

    /* start the resource manager message loop */
    while(1) {
        if((ctp = dispatch_block(ctp)) == NULL) {
            fprintf(stderr, "block error\n");
            return EXIT_FAILURE;
        }
        dispatch_handler(ctp);
    }
}

int
io_read (resmgr_context_t *ctp, io_read_t *msg, RESMGR_OCB_T *ocb)
{
    int         nleft;
    int         nbytes;
    int         nparts;
    int         status;

    if ((status = iofunc_read_verify (ctp, msg, ocb, NULL)) != EOK)
        return (status);
        
    if ((msg->i.xtype & _IO_XTYPE_MASK) != _IO_XTYPE_NONE)
        return (ENOSYS);

    /*
     *  On all reads (first and subsequent), calculate
     *  how many bytes we can return to the client,
     *  based upon the number of bytes available (nleft)
     *  and the client's buffer size
     */

    nleft = ocb->attr->nbytes - ocb->offset;
    nbytes = min (msg->i.nbytes, nleft);

    if (nbytes > 0) {
        /* set up the return data IOV */
        SETIOV (ctp->iov, buffer + ocb->offset, nbytes);

        /* set up the number of bytes (returned by client's read()) */
        _IO_SET_READ_NBYTES (ctp, nbytes);

        /*
         * advance the offset by the number of bytes
         * returned to the client.
         */

        ocb->offset += nbytes;
        
        nparts = 1;
    } else {
        /*
         * they've asked for zero bytes or they've already previously
         * read everything
         */
        
        _IO_SET_READ_NBYTES (ctp, 0);
        
        nparts = 0;
    }

    /* mark the access time as invalid (we just accessed it) */

    if (msg->i.nbytes > 0)
        ocb->attr->flags |= IOFUNC_ATTR_ATIME;

    return (_RESMGR_NPARTS (nparts));
}

上述代碼中，ocb結(jié)構(gòu)存放了緩存的偏移位置疾就，此外指向的屬性結(jié)構(gòu)體attr中澜术，存放了緩存的實(shí)際大小。而attr中緩沖的大小是通過iofunc_attr_init()接口設(shè)置進(jìn)去的猬腰。
此外鸟废，在代碼中需要注意用到了ctp->iov，使用SETIOV()宏姑荷，將ctp->iov指向了需要回復(fù)的數(shù)據(jù)盒延。
當(dāng)沒有讀取到數(shù)據(jù)時(shí)缩擂，返回(_RESMGR_NPARTS(0))，當(dāng)讀取到數(shù)據(jù)后兰英，怎么來告知客戶端讀取了多少呢撇叁？代碼中_IO_SET_READ_NBYTES()宏將讀取的信息保存到了ctp中，當(dāng)返回到庫的時(shí)候畦贸，會(huì)將讀取的字節(jié)數(shù)當(dāng)成MsgReplyv()函數(shù)的參數(shù)陨闹，并通知內(nèi)核MsgSend()該發(fā)送什么，最終上層便能獲取到字節(jié)數(shù)薄坏。

2. Handling the _IO_WRITE message

io_write處理函數(shù)負(fù)責(zé)處理_IO_WRITE消息趋厉，將數(shù)據(jù)寫入到設(shè)備。當(dāng)客戶端調(diào)用write()/fflush()等操作時(shí)執(zhí)行胶坠。
消息定義如下：

struct _io_write {
    uint16_t            type;
    uint16_t            combine_len;
    int32_t             nbytes;
    uint32_t            xtype;
    /* unsigned char    data[nbytes]; */
};

typedef union {
    struct _io_write    i;
    /*  nbytes is returned with MsgReply  */
} io_write_t;

這個(gè)結(jié)構(gòu)中的字段與io_read_t中一致君账，下邊看一個(gè)寫操作的代碼：

int
io_write (resmgr_context_t *ctp, io_write_t *msg, RESMGR_OCB_T *ocb)
{
    int     status;
    char    *buf;

    if ((status = iofunc_write_verify(ctp, msg, ocb, NULL)) != EOK)
        return (status);

    if ((msg->i.xtype & _IO_XTYPE_MASK) != _IO_XTYPE_NONE)
        return(ENOSYS);

    /* set up the number of bytes (returned by client's write()) */

    _IO_SET_WRITE_NBYTES (ctp, msg->i.nbytes);

    buf = (char *) malloc(msg->i.nbytes + 1);
    if (buf == NULL)
        return(ENOMEM);

    /*
     *  Reread the data from the sender's message buffer.
     *  We're not assuming that all of the data fit into the
     *  resource manager library's receive buffer.
     */

    resmgr_msgread(ctp, buf, msg->i.nbytes, sizeof(msg->i));
    buf [msg->i.nbytes] = '\0'; /* just in case the text is not NULL terminated */
    printf ("Received %d bytes = '%s'\n", msg -> i.nbytes, buf);
    free(buf);

    if (msg->i.nbytes > 0)
        ocb->attr->flags |= IOFUNC_ATTR_MTIME | IOFUNC_ATTR_CTIME;

    return (_RESMGR_NPARTS (0));
}

當(dāng)在寫入的時(shí)候，如果提供的buffer大小不夠沈善，而需要寫入的數(shù)據(jù)又很大乡数，此時(shí)就需要使用一個(gè)循環(huán)機(jī)制來多次寫入了。

3. Methods of return and replying

可以使用各種方式從處理程序中返回到資源管理器庫：

• 返回錯(cuò)誤
  比如申請(qǐng)內(nèi)存不夠時(shí)：return (ENOMEM)

• 返回指向數(shù)據(jù)的IOV數(shù)組
  可以使用IOV數(shù)組來指向多片數(shù)據(jù)闻牡，比如如下代碼：

my_header_t     header;
a_buffer_t      buffers[N];

...

SETIOV(&ctp->iov[0], &header, sizeof(header));
SETIOV(&ctp->iov[1], &buffers[i], sizeof(buffers[i]));
return (_RESMGR_NPARTS(2));

• 返回一個(gè)包含數(shù)據(jù)的緩存
  比如在響應(yīng)read()請(qǐng)求時(shí)净赴，所有的數(shù)據(jù)都在一個(gè)緩存中，那么可以以下兩種方式：

return (_RESMGR_PTR(ctp, buffer, nbytes));

和：

SETIOV (ctp->iov, buffer, nbytes);
return (_RESMGR_NPARTS(1));

• 成功返回罩润，但是不攜帶數(shù)據(jù)
  比如：return (EOK)玖翅，不過更常見的方式是：return (_RESMGR_NPARTS(0))。

• 讓資源管理器庫去返回

• 在服務(wù)器端執(zhí)行回復(fù)
  上邊講到的返回情況割以，都是通過資源管理器庫調(diào)用MsgReply*()/MsgError()來解除客戶端的阻塞狀態(tài)金度，在有些case中，你可能不需要這些庫來回復(fù)严沥，可以使用return (_RESMGR_NOREPLY)

• 推遲回復(fù)猜极，讓客戶端保持阻塞
  一個(gè)例子是管道資源管理器，當(dāng)一個(gè)客戶端讀取管道時(shí)消玄，而此時(shí)管道中沒有數(shù)據(jù)魔吐，可以有兩種選擇：1）返回錯(cuò)誤，2）保持客戶端阻塞莱找，等有數(shù)據(jù)時(shí)再回復(fù)酬姆。
  另外一個(gè)例子是往一個(gè)設(shè)備寫數(shù)據(jù)時(shí)，希望數(shù)據(jù)全部寫入后再回復(fù)奥溺。

• 返回并告訴庫執(zhí)行默認(rèn)處理
  可以執(zhí)行return (_RESMGR_DEFAULT)辞色。

4. Handling other read/write details

4.1 處理xtype成員

從上文中可知，在io_read_t/io_write_t等消息結(jié)構(gòu)中浮定，有一個(gè)xtype成員相满，這個(gè)成員包含了可用于調(diào)整標(biāo)準(zhǔn)I/O函數(shù)行為的一些擴(kuò)展信息层亿，如下：

• _IO_XTYPE_NONE，沒有提供擴(kuò)展類型信息立美；
• _IO_XTYPE_OFFSET匿又，客戶端調(diào)用pread()/pread64()/pwrite()/pwrite64()時(shí)，不希望用到OCB中的偏移建蹄，而是提供了一個(gè)一次性的偏移量碌更，該偏移量存放在消息緩沖的開頭，緊跟在struct _io_read或struct _io_write后洞慎，比如：

struct myread_offset {
    struct _io_read        read;
    struct _xtype_offset   offset;
}   

• _IO_XTYPE_READCOND痛单，如果客戶端調(diào)用readcond()，通常會(huì)對(duì)時(shí)間和緩沖區(qū)的大小增加一些限制劲腿。這個(gè)限制放置在消息緩沖區(qū)的開頭旭绒，緊跟在struct _io_read或struct _io_write后，比如：

struct myreadcond {
    struct _io_read        read;
    struct _xtype_readcond cond;
}   

• _IO_XFLAG_DIR_EXTRA_HINT焦人，只有在讀取目錄時(shí)才有效挥吵。

當(dāng)你不想使用擴(kuò)展功能時(shí)，可以參考以下例子：

int
io_read (resmgr_context_t *ctp, io_read_t *msg,
         RESMGR_OCB_T *ocb)
{
    int    status;

    if ((status = iofunc_read_verify(ctp, msg, ocb, NULL))
         != EOK) {
        return (status);
    }

    /* No special xtypes */
    if ((msg->i.xtype & _IO_XTYPE_MASK) != _IO_XTYPE_NONE)
        return (ENOSYS);

    ...
}

4.2 處理pread*()和pwrite*()

客戶端調(diào)用pread*()函數(shù)時(shí)花椭，處理_IO_READ消息的示例代碼如下：

/* we are defining io_pread_t here to make the code below
   simple */
typedef struct {
    struct _io_read         read;
    struct _xtype_offset    offset;
} io_pread_t;

int
io_read (resmgr_context_t *ctp, io_read_t *msg,
         RESMGR_OCB_T *ocb)
{
    off64_t offset; /* where to read from */
    int     status;

    if ((status = iofunc_read_verify(ctp, msg, ocb, NULL))
         != EOK) {
        return(status);
    }
    
    switch(msg->i.xtype & _IO_XTYPE_MASK) {
    case _IO_XTYPE_NONE:
        offset = ocb->offset;
        break;
    case _IO_XTYPE_OFFSET:
        /*
         *  io_pread_t is defined above.
         *  Client is doing a one-shot read to this offset by
         *  calling one of the pread*() functions
         */
        offset = ((io_pread_t *) msg)->offset.offset;
        break;
    default:
        return(ENOSYS);
    }

    ...
}

客戶端調(diào)用pwrite*()時(shí)忽匈，處理_IO_WRITE消息的示例代碼如下：

/* we are defining io_pwrite_t here to make the code below
   simple */
typedef struct {
    struct _io_write        write;
    struct _xtype_offset    offset;
} io_pwrite_t;

int
io_write (resmgr_context_t *ctp, io_write_t *msg,
          RESMGR_OCB_T *ocb)
{
    off64_t offset; /* where to write */
    int     status;
    size_t  skip;   /* offset into msg to where the data
                       resides */

    if ((status = iofunc_write_verify(ctp, msg, ocb, NULL))
         != EOK) {
        return(status);
    }
    
    switch(msg->i.xtype & _IO_XTYPE_MASK) {
    case _IO_XTYPE_NONE:
        offset = ocb->offset;
        skip = sizeof(io_write_t);
        break;
    case _IO_XTYPE_OFFSET:
        /* 
         *  io_pwrite_t is defined above
         *  client is doing a one-shot write to this offset by
         *  calling one of the pwrite*() functions
         */
        offset = ((io_pwrite_t *) msg)->offset.offset;
        skip = sizeof(io_pwrite_t);
        break;
    default:
        return(ENOSYS);
    }

    ...
    
    /* 
     *  get the data from the sender's message buffer, 
     *  skipping all possible header information
     */
    resmgr_msgreadv(ctp, iovs, niovs, skip);
    
    ...
}

需要注意的是，通常情況下在發(fā)送消息緩沖中个从，數(shù)據(jù)會(huì)緊跟著struct _io_write脉幢，而在上邊這種情況下歪沃，數(shù)據(jù)緊跟在struct _xtype_offset后邊嗦锐，這里邊會(huì)存在一個(gè)偏移。

4.3 處理readcond()

與處理pread()/_IO_XTYPE_OFFSET類似沪曙，如下：

typedef struct {
    struct _io_read        read;
    struct _xtype_readcond cond;
} io_readcond_t

struct _xtype_readcond *cond
...
    CASE _IO_XTYPE_READCOND:
        cond = &((io_readcond_t *)msg)->cond
        break;
}

5. Updating the time for reads and writes

在讀操作的那個(gè)示例代碼中奕污，有如下代碼：

if (msg->i.nbytes > 0)
        ocb->attr->flags |= IOFUNC_ATTR_ATIME;

根據(jù)POSIX標(biāo)準(zhǔn)，當(dāng)讀取大于0字節(jié)的數(shù)據(jù)并且成功后液走，需要去更新訪問時(shí)間碳默，但是POSIX標(biāo)準(zhǔn)中并沒有說需要立刻去更新，因此當(dāng)有多次讀的時(shí)候缘眶，可能并不想每次都從內(nèi)核中獲取時(shí)間嘱根，可以在需要更新的時(shí)候再統(tǒng)一更新，這樣做的缺點(diǎn)是時(shí)間記錄的不是每一次讀操作的時(shí)間巷懈。如果想記錄每次讀的時(shí)間该抒，那么可以調(diào)用iofunc_time_update()接口，代碼如下：

if (msg->i.nbytes > 0) {
        ocb->attr->flags |= IOFUNC_ATTR_ATIME;
        iofunc_time_update(ocb->attr);
}

寫操作也是一樣的原理顶燕。