pipe.c

Sun公司Dream项目

/*
 * The contents of this file are subject to the terms
 * of the Common Development and Distribution License
 * (the "License").  You may not use this file except
 * in compliance with the License.
 *
 * You can obtain a copy of the license at
 * http://www.opensource.org/licenses/cddl1.php
 * See the License for the specific language governing
 * permissions and limitations under the License.
 *
 * When distributing Covered Code, include this CDDL
 * HEADER in each file and include the License file at
 * http://www.opensource.org/licenses/cddl1.php.  If 
 * applicable, add the following below this CDDL HEADER, 
 * with the fields enclosed by brackets "[]" replaced 
 * with your own identifying information: 
 * Portions Copyright [yyyy]
 * [name of copyright owner]
 */ 

/*
 * $(@)Pipe.c $Revision: 1.2 $ $Date: 2006/07/15 00:02:39 $
 * 
 * Copyright 2006 Sun Microsystems, Inc. All Rights Reserved.
 */
/*
 * Copyright (c) 1996 by Sun Microsystems, Inc.
 * All rights reserved.
 *
 * This software is the confidential and proprietary information of Sun
 * Microsystems, Inc. ("Confidential Information").  You shall not
 * disclose such Confidential Information and shall use it only in
 * accordance with the terms of the license agreement you entered into
 * with Sun.
 */

/*
 * Pipe.c -- Description of Pipe
 *
 * This file generated from Tmplate version: 1.4 99/10/27
 * 
 * A Pipe is a object that passes data from a master thread to a slave thread in
 * a co-routine manner.  The master "puts" data to the Pipe and then gives up
 * control to the slave while the slave processes the passed data, when the
 * slave has consumed the data control returns to the master.  The slave
 * accesses data by requesting pointers to a contiguous chunk of data, the
 * max size of that chunk is specified at the time the Pipe in created.  Pipe
 * correctly handles collecting chunks when they break across master puts.
 * The pointer returned by Pipe to a slave and the data it points at is only
 * valid until the next call to Pipe, therefore, the slave should copy the
 * pointed to data before it's next call to get data.
 * 
 * The master indicates end of data to the slave by calling pipeEof(). A slave
 * read after pipeEof() will cause the slave to longjmp to an eof handler.
 * 
 * The slave should NEVER call thr_exit().
 */


#pragma ident "@(#)Pipe.c 1.8	99/12/10 SMI"

#if	!defined(PIPE_HEADER)
#define	PIPE_BODY
#define	PIPE_INLINE		extern
#include "mmp/Pipe.h"
#endif					   /* !defined(PIPE_HEADER) */

/***********************************************************************
 * Includes
 ***********************************************************************/

#ifdef	ASSERTS
#include <errno.h>
#endif	/* ASSERTS */
#include <setjmp.h>
#include <thread.h>
#include <synch.h>
#include <string.h>
#include <stdlib.h>
#include <stdio.h>

#include "cobjs/Foreign.h"
#include "cobjs/Inline.h"
#include "cobjs/Log.h"
#include "cobjs/Macros.h"
#include "cobjs/Types.h"
#include "cobjs/RetCode.h"

#include "mmp/Mmp.h"
#include "mmp/Crc32Table.h"

/*************************************************************************
 * Defines
 *************************************************************************/

#define	THR_BKG_BUG		0	/* job in bkg causes lwp starvation */

#define	PIPE_MAX_SEGMENTS	128
#define	N_BUF_POS		4	/* expect < 4 discontig segs residue */

/*************************************************************************
 * Instance Variables
 *************************************************************************/

typedef struct BufPosList {
    size_t		len;
    PipePosition	position;
} BufPosList;

typedef struct Buf {
    u8                 *basep;		/* base of buffer */
    size_t              len;		/* if alloc'ed, size of buf, else 0 */
    u8                 *outp;		/* current out pointer */
    size_t              avail;		/* chars in buffer */
    size_t              marked;		/* marked chars in buf */
    BufPosList		*bufPosList;	/* pos in orig stream */
    int			bplGet;		/* cur get entry in bufPosList */
    int			bplPut;		/* cur put entry in bufPosList */
    int			bplSize;	/* size of bufPosList */
} Buf;

typedef enum PipeThreadState {
    PIPETHREADSTATE_INIT,		/* hold til master init done */
    PIPETHREADSTATE_RUNNING,		/* parsing */
    PIPETHREADSTATE_ZOMBE		/* error, waiting to free */
} PipeThreadState;

struct _Pipe {
    Buf                 residueBuf;
    Buf                 masterBufs[PIPE_MAX_SEGMENTS];
    int			curSeg;
    int			nSegs;
    u32                 usedBits;

    PipePosition        position;	    /* pipe relative position */

    Boolean             doCrc;
    u32                 crcSum;

    int                *eofJmpBuf;
    RetCode             retCode;

    PipeParser          parser;
    PipeCleanup         cleanup;
    void               *instp;
    void               *cop;
    u16                 pid;

    Pipe                lastDstPipe;
    Boolean		isFlush;
    Boolean		isEof;

    PipeThreadState     threadState;
    sema_t              masterSema;
    sema_t              slaveSema;
    thread_t            threadId;
};

/*************************************************************************
 * Private types and prototypes referenced from inlines
 *************************************************************************/

/*
 * Use INLINE_PRIVATE if non-inline-able, define in Non-inlinable section
 * Use static if inline-able, define in Private Inline-able section
 *
 * INLINE_PRIVATE void pipeInlinePrivate(void);
 */
static Buf         *pipeFillBuf(Pipe pipe, size_t size);
static void         pipeCrcSum(Pipe pipe, u8 *datap, int len);
static u8          *bufGet(Buf *bufp, size_t len);

#ifdef __cplusplus
extern "C" {
#endif

INLINE_PRIVATE Buf * pipeFillBuf2(Pipe pipe, size_t size);

#ifdef __cplusplus
}
#endif

/*************************************************************************
 * Private class data referenced from inlines
 *************************************************************************/

/*************************************************************************
 * Inline Methods
 *************************************************************************/

/*
 * PIPE_INLINE void *
 * pipeInlineRoutine(void)
 * {
 * }
 */

PIPE_INLINE u8 *
pipeGet(Pipe pipe, size_t size)
{
    u8                 *p;
    Buf		       *bufp;

    ASSERT(pipe->usedBits == 0);
    if ((bufp = pipeFillBuf(pipe, size)) == NULL) {
	longjmp(pipe->eofJmpBuf, 1);
    }
    pipe->position += size;
    p = bufGet(bufp, size);
    if (pipe->doCrc) {
	pipeCrcSum(pipe, p, size);
    }
    return p;
}

PIPE_INLINE u8 *
pipePeek(Pipe pipe, size_t size)
{
    Buf		       *bufp;

    ASSERT(pipe->usedBits == 0);
    if ((bufp = pipeFillBuf(pipe, size)) == NULL) {
	longjmp(pipe->eofJmpBuf, 1);
    }
    return bufp->outp;
}

PIPE_INLINE void
pipeSkip(Pipe pipe, size_t size)
{
    size_t              osize = size;
    size_t              rem;
    u8                 *p;

    ASSERT(pipe->usedBits == 0);
    while (size != 0) {
        Buf		       *bufp;

	if ((bufp = pipeFillBuf(pipe, 1)) == NULL) {
	    longjmp(pipe->eofJmpBuf, 1);
	}

	rem = MIN(size, bufp->avail);
	size -= rem;

	p = bufGet(bufp, rem);
	if (pipe->doCrc) {
	    pipeCrcSum(pipe, p, rem);
	}
    }
    pipe->position += osize;
}

/*************************************************************************
 * Private Inlineable Methods and Functions Called From Inlines
 *************************************************************************/

static Buf         *
pipeFillBuf(Pipe pipe, size_t size)
{
    return pipe->isEof ? NULL
	: (pipe->residueBuf.avail != 0
	  || pipe->masterBufs[pipe->curSeg].avail < size)
		? pipeFillBuf2(pipe, size)
	: &pipe->masterBufs[pipe->curSeg];
}

#if	0
/**
 *            32   26   23   22   16   12   11   10   8   7   5   4   2
 * Calculate x  + x  + x  + x  + x  + x  + x  + x  + x + x + x + x + x + x + 1
 * 
 * 3 3 2 2  2 2 2 2  2 2 2 2  1 1 1 1  1 1 1 1  1 1 0 0  0 0 0 0  0 0 0 0
 * 1 0 9 8  7 6 5 4  3 2 1 0  9 8 7 6  5 4 3 2  1 0 9 8  7 6 5 4  3 2 1 0
 * -------  -------  -------  -------  -------  -------  -------  -------
 * 0 0 0 0  0 1 0 0  1 1 0 0  0 0 0 1  0 0 0 1  1 1 0 1  1 0 1 1  0 1 1 1
 *       0        4        c        1        1        d        b        7
 */

static void
pipeCrcSum(Pipe pipe, u8 *datap, int len)
{
    /*
     * Direct calculation code
     */
    while (len-- > 0) {
	int                 bit;
	u32                 input = *datap++ << 24;

	for (bit = 7; bit >= 0; bit -= 1) {
	    s32                 newZ0 = input ^ pipe->crcSum;

	    pipe->crcSum <<= 1;
	    if (newZ0 < 0) {
		pipe->crcSum ^= 0x04c11db7;
	    }
	    input <<= 1;
	}
    }
}
#endif	/* 0 */

static void
pipeCrcSum(Pipe pipe, u8 *datap, int len)
{
    /*
     * Table-based calculation code
     */
    while (len-- > 0) {
	pipe->crcSum = (pipe->crcSum >> 8)
	    ^ crc32Table[(pipe->crcSum ^ *datap++) & 0xff];
    }
}

static u8          *
bufGet(Buf *bufp, size_t size)
{
    u8                 *p = bufp->outp;

    ASSERT(size <= bufp->avail);
    if (size != 0) {
	bufp->outp += size;
	bufp->avail -= size;
	if (bufp->avail < bufp->marked) {
	    bufp->marked = 0;
	}
	while (size != 0) {
	    BufPosList *bplp = &bufp->bufPosList[bufp->bplGet];
	    size_t chunk = MIN(size, bplp->len);
	    ASSERT(bufp->bplGet < bufp->bplPut);
	    ASSERT(bplp->len > 0);
	    bplp->len -= chunk;
	    bplp->position += chunk;
	    if (bplp->len == 0) {
		bufp->bplGet += 1;
	    }
	    size -= chunk;
	}
    }
    return p;
}

#if	!defined(PIPE_HEADER)

/*************************************************************************
 * Private types
 *************************************************************************/

/*************************************************************************
 * Private method prototypes
 *************************************************************************/

static RetCode      pipePut2(Pipe pipe, void *buf, size_t len, Boolean isMarked,
			Boolean doFlush, PipePosition bufPosition);
static RetCode      pipeFlush(Pipe pipe, Boolean isForced);
static Boolean	    pipeIsPushBack(Pipe pipe);

/*************************************************************************
 * Private function prototypes
 *************************************************************************/

static void        *pipeThreadWrapper(void *arg);

static Buf	    bufInit(u8 *buf, size_t len, int bplSize);
static void	    bufAssign(Buf *bufp, u8 *datap, size_t avail,
			Boolean isMarked, PipePosition bufPosition);
static void         bufTransfer(Buf *destBufp, Buf *srcBufp, size_t len);
static Boolean      bufIsAtMark(Buf *bufp);
static Boolean      bufIsEmpty(Buf *bufp);

/*************************************************************************
 * Private class data
 *************************************************************************/

static u32          bitMask[] = {
    0x00000000, 0x00000001, 0x00000003, 0x00000007,
    0x0000000f, 0x0000001f, 0x0000003f, 0x0000007f,
    0x000000ff, 0x000001ff, 0x000003ff, 0x000007ff,
    0x00000fff, 0x00001fff, 0x00003fff, 0x00007fff,
    0x0000ffff, 0x0001ffff, 0x0003ffff, 0x0007ffff,
    0x000fffff, 0x001fffff, 0x003fffff, 0x007fffff,
    0x00ffffff, 0x01ffffff, 0x03ffffff, 0x07ffffff,
    0x0fffffff, 0x1fffffff, 0x3fffffff, 0x7fffffff,
    0xffffffff
};

static RetCodeTable errorTable[] = {
    {PIPE_ERROR_EOF, NULL, "unexpected eof"},
    {PIPE_ERROR_STUFF_BYTE, NULL, "invalid stuff byte"},
    {PIPE_ERROR_EXTRANEOUS_DATA, NULL, "extraneous data at end of stream"},
    {0, NULL, NULL}