dbbufferc.nc

nesC写的heed算法

// $Id: DBBufferC.nc,v 1.11.4.6 2003/09/06 00:34:18 whong Exp $

/*									tab:4
 * "Copyright (c) 2000-2003 The Regents of the University  of California.  
 * All rights reserved.
 *
 * Permission to use, copy, modify, and distribute this software and its
 * documentation for any purpose, without fee, and without written agreement is
 * hereby granted, provided that the above copyright notice, the following
 * two paragraphs and the author appear in all copies of this software.
 * 
 * IN NO EVENT SHALL THE UNIVERSITY OF CALIFORNIA BE LIABLE TO ANY PARTY FOR
 * DIRECT, INDIRECT, SPECIAL, INCIDENTAL, OR CONSEQUENTIAL DAMAGES ARISING OUT
 * OF THE USE OF THIS SOFTWARE AND ITS DOCUMENTATION, EVEN IF THE UNIVERSITY OF
 * CALIFORNIA HAS BEEN ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 * 
 * THE UNIVERSITY OF CALIFORNIA SPECIFICALLY DISCLAIMS ANY WARRANTIES,
 * INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY
 * AND FITNESS FOR A PARTICULAR PURPOSE.  THE SOFTWARE PROVIDED HEREUNDER IS
 * ON AN "AS IS" BASIS, AND THE UNIVERSITY OF CALIFORNIA HAS NO OBLIGATION TO
 * PROVIDE MAINTENANCE, SUPPORT, UPDATES, ENHANCEMENTS, OR MODIFICATIONS."
 *
 * Copyright (c) 2002-2003 Intel Corporation
 * All rights reserved.
 *
 * This file is distributed under the terms in the attached INTEL-LICENSE     
 * file. If you do not find these files, copies can be found by writing to
 * Intel Research Berkeley, 2150 Shattuck Avenue, Suite 1300, Berkeley, CA, 
 * 94704.  Attention:  Intel License Inquiry.
 */
#ifdef kMATCHBOX
includes Matchbox;
#endif

#ifndef NUMBUFFERS
#define NUMBUFFERS	4
#endif

module DBBufferC {
  uses {
    interface RadioQueue; //writes results out to a radio queue
    interface MemAlloc;
    interface QueryProcessor;
    interface Leds;
    interface CommandUse;
    interface QueryResultIntf;
    interface ParsedQueryIntf;
    interface TupleIntf;
    interface AttrUse;
#ifdef kMATCHBOX
    interface FileWrite;
    interface FileWrite as HeaderFileWrite;
    interface FileRead;
    interface FileRead as HeaderFileRead;
    interface FileRename;
    interface FileDelete;
    interface FileDir;
#endif
#ifdef kUART_DEBUGGER
    interface Debugger as UartDebugger;
#endif

    command void allocDebug();
  } 
  provides {
    interface DBBuffer;
    interface StdControl;
    interface CatalogTable;
#ifdef kMATCHBOX
    event result_t fsReady();
#endif
  }
}

#ifdef kUART_DEBUGGER
#define DEBUG(s,l)  (TOS_LOCAL_ADDRESS==0?l:(call UartDebugger.writeLine((s),(l))))
#else
#define DEBUG(s,l)
#endif

/**
<p>
   DBBuffer is a output buffer abstraction for TinyDB.  
   Buffers can be one of several types:
<p>
   Radio:  Results are written directly out to the radio
   Command:  Each result causes a command to be invoked
   RAM:  Results are written to a RAM buffer
   EEPROM:  Results are written to EEPROM
   Catalog buffers (e.g. event, attribute, command, query) that are read-only 
    and contain system meta-data.
<p>
   RAM and EEPROM buffers use a cicrular buffer, with nextFree and
   nextFull slots if nextFull slot is after nextFree slot, then used
   slots are from nextFull to the end of the buffer plus.  Otherwise,
   used slots are from nextFull to nextFree.
<p>
   To the outside world, buffer appears to have a number of used
   slots, with the first slot (0) beginning at nextFull.
<p>
<code>	 		
       nextFull > nextFree       	       	   nextFull < nextFree
       +--------------------------+ top	           +--------------------------+ top
       |..........................|                |                          |	   
       |.....used slots...........|                |      free slots          |	   
       |..........................|                |                          |	   
       |..........................|                |                          |	   
       +--------------------------+ nextFree       +--------------------------+	nextFull
       |                          |                |..........................|		
       |                          |                |..........................|                        	
       |                          |                |..........................|		
       |      free slots          |                |.......used slots.........|		
       |                          |                |..........................|		
       |                          |                |..........................|		
       |                          |                |..........................|		
       +--------------------------+ nextFull       +--------------------------+ nextFree
       |..........................|                |                          |	       	
       |.....used slots...........|                |                          |
       |..........................|                |       free slots         |
       |..........................|                |                          |
       +--------------------------+                +--------------------------+
</code>
*/

implementation {
  enum {
    kNUM_BUFS = NUMBUFFERS,
    kATTR_BUF = kNUM_BUFS + 1,
    kCOMMAND_BUF = kATTR_BUF + 1,
    kEVENT_BUF = kCOMMAND_BUF + 1,
    kQUERY_BUF = kEVENT_BUF + 1
  };

  typedef enum {
    kIDLE,
    kALLOC_SCHEMA,
    kALLOC_BUFFER,
    kALLOC_ROW,
    kALLOC_NAME,
    kREAD_ROW,
    kEEPROM_WRITE,
    kEEPROM_ALLOC_ROW,
    kREADING_LENGTH,   //stages in reading of data from EEPROM
    kALLOC_FOR_READ,
    kREADING_DATA,
    kWRITE_BUFFER,
    kHEADER_READ,
    kWRITE_FILE_COUNT, 
    kREAD_FILE_COUNT,
    kREAD_OPEN,
    kREAD_LENGTHS,
    kSKIP_BYTES,
    kREAD_NAME,
    kREAD_FIELD_LEN,
    kALLOC_FIELD_DATA,
    kREAD_FIELD_DATA,
    kALLOC_QUERY,
    kREAD_QUERY,
    kREAD_BUFFER,
    kWRITING_LENGTHS,
    kWRITING_NAME,
    kWRITING_QUERY,
    kWRITING_BUFFER,
    kWRITE_FIELD_LEN,
    kWRITE_FIELD_DATA,
    kWRITE_NEXT_BUFFER,
    kNOT_READY,
    kOPEN_BUFFER,
    kRADIO_ENQUEUE,
    kCLEANUP
 } AllocState;


  typedef struct {
    BufferType type;
    BufferPolicy policy;
    ParsedQuery **qh; //schema of buffer (rest of query fields ignored)
    uint16_t numRows; //number of rows in buffer
    uint16_t nextFree; //next free row
    uint16_t nextFull; //next full row (top of queue)

    uint16_t len; //length, in bytes, of buffers
    uint16_t rowSize; //length, in bytes, of one row

    long data;

    char **name;

    union {
      char **bufdata; //data handle for RAM buffers
      struct {
	bool isOpen; //has the eeprom been successfully opened
	bool isWrite;
      } eeprom;
    } u;
  } Buffer;

  Buffer mBuffers[kNUM_BUFS];
  long mUsedBufs;
  uint8_t mCurBuffer;
  uint16_t mCurRow;
  uint8_t mLen;
  uint8_t mSearchFile;
  uint8_t mCurFile;
  uint8_t mLenBytes[3];
  
  Handle mCurRowHandle;
  Handle mEepromRow;
  QueryResultPtr mCurResult;
  ParsedQuery *mAllocSchema, *mCurQuery;
  uint8_t mCurResultIdx;
  AllocState mState;

  char *mNamePtr;

  //#if defined(PLATFORM_PC)
  //#ifdef kDEBUG
  //#endif
#ifdef kDEBUG
  ParsedQueryPtr mQueryPtr;
  QueryResult mQr;
  short mInt;
  char tupleBuf[50];
  char queryBuf[100];
  char aggResult[7];
#endif

  char mDbgBuf[10];

  //static schemas for catalog buffers
  enum {NUM_ATTR_FIELDS = 1};
  uint8_t ATTR_TYPES[NUM_ATTR_FIELDS];
  uint8_t ATTR_SIZES[NUM_ATTR_FIELDS];
  char ATTR_NAMES[NUM_ATTR_FIELDS][STRING_SIZE];


  TinyDBError calcNextFreeRow(Buffer *buf);
  TinyDBError getBuf(uint8_t bufId, Buffer **buf);
  void cleanupBuffer(int i, int lineno);
  void initCatalog();
			
  TinyDBError getSpecialBufferId(BufferType type, uint8_t *bufferId);
  TinyDBError continueRadioBufferEnqueue();

#ifdef kMATCHBOX
  uint8_t mNumFiles;
  bool mLocalRead;
  Handle mNameHand;
  Handle mQueryHandle;
  AllocState mHeaderFileState, mAppendState;
  char *mCurBufName;
  uint8_t mCurBufId;
  Buffer *mReplaceBuf, *mAppendBuf;
  uint8_t mReplaceBufId;
  uint8_t mCurWriteIdx;
  bool mFoundBuffer;
  uint8_t mNumSkipBytes;
  bool mNameSearch;
  char mReadBuf[10];
  bool mDelete;
  char mCurName[15];
  bool mHasName;
  bool mAllocing;
  bool mDidFieldLen;

  TinyDBError doLoad();
  void headerReadFailed(TinyDBError err);
  TinyDBError writeFileCount();
  void writeCountDone(TinyDBError err);
  TinyDBError readFileCount();
  void readCountDone(TinyDBError err);
  void readBuffer(char *bufName, uint8_t bufId, TinyDBError err);

  task void loadBufferTask();
  task void appendBufferTask();
  task void replaceBufferTask();
  task void readEEPROMRow();

#define kVERSION_CODE 0x01 //the version of the DBBuffer 

#endif

  /* ----------------------------------- StdControl Methods ------------------------------------- */

  command result_t StdControl.init() {
    mUsedBufs = 1; //clear free bitmap -- can't use first buffer -- it's for the radio!
#ifdef kMATCHBOX
    mState = kNOT_READY; //wait until the EEPROM file system is ready
    mNumFiles = 0;
    mAllocing = FALSE;
    mHeaderFileState = kIDLE;
#else
    mState = kIDLE;
#endif
#ifdef kDEBUG
    mInt = 0;
#endif
    initCatalog();
    mEepromRow = NULL;


    
    return SUCCESS;
  }

  command result_t StdControl.start() {
#ifdef kDEBUG
    BufInfo buf;
    bool pending;
    TinyDBError err;
    Expr e;

    mQueryPtr = (ParsedQueryPtr)queryBuf;

    dbg(DBG_USR2, "start"); fflush(stdout);

    mQueryPtr->qid = 0;
    mQueryPtr->numFields = 1;
    mQueryPtr->numExprs = 1;
    mQueryPtr->epochDuration = 1024;
    mQueryPtr->clocksPerSample = 1024/32;
    mQueryPtr->clockCount = mQueryPtr->clocksPerSample;
    mQueryPtr->currentEpoch = 0;
    mQueryPtr->fromQid = kNO_QUERY;
    mQueryPtr->bufferType = kRAM;
    mQueryPtr->queryToSchemaFieldMap[0] = 1; //some field in the schema?
    memset(aggResult, 0, 7);
    aggResult[4]=10;
    aggResult[5]=10;
    aggResult[6]=10;
    
    e.opType = kAGG;
    e.idx = 0;
    e.fieldOp = FOP_NOOP;
    e.fieldConst = 0;
    e.ex.agg.field = 0;
    e.ex.agg.groupingField = kNO_GROUPING_FIELD;
    e.ex.agg.groupFieldOp = FOP_NOOP;
    e.ex.agg.groupFieldConst = 0;
    e.ex.agg.op = MIN;

    call ParsedQueryIntf.setExpr(mQueryPtr, 0, e);
    err = call DBBuffer.nextUnusedBuffer(&mQueryPtr->bufferId);
    if (err != err_NoError)
      dbg(DBG_USR2, "err, nextUnusedBuf: %d", err);
    buf.ram.numRows = 2;
    buf.ram.policy = EvictOldestPolicy;
    
    mQueryPtr->buf = buf;

    err = call DBBuffer.alloc(mQueryPtr->bufferId, kRAM, buf.ram.numRows, buf.ram.policy,&mQueryPtr, &pending, 0);
    if (err != err_NoError)
      dbg(DBG_USR2, "err, DBBuffer.alloc: %d", err);
    
#endif
      return SUCCESS;
  }

  command result_t StdControl.stop() {
      return SUCCESS;
  }


  void initCatalog() {
    ATTR_TYPES[0] = STRING;
    ATTR_SIZES[0] = sizeOf(STRING);
    strcpy(ATTR_NAMES[0],"name");
  }

#ifdef kDEBUG
  void enqueueResult() {
    bool pending;
    TinyDBError err;


    mInt++;

    dbg(DBG_USR2, "enqueuing"); fflush(stdout);

    err = call QueryResultIntf.initQueryResult(&mQr);
    if (err != err_NoError)
      dbg(DBG_USR2, "err, initQueryResult: %d", err);

    err = call QueryResultIntf.addAggResult(&mQr, 0, aggResult, 7, mQueryPtr, 0);
    if (err != err_NoError)
      dbg(DBG_USR2, "err, addAggResult: %d", err);

    err = call DBBuffer.enqueue(mQueryPtr->bufferId, &mQr, &pending, mQueryPtr);
    if (err != err_NoError)
      dbg(DBG_USR2, "err, enqueue: %d", err);


  }


  void readResult() {
    QueryResult qr;
    ParsedQuery *pq;
    TinyDBError err;
    short result;

    bool pending;
    Tuple *t;

    err = call DBBuffer.peek(mQueryPtr->bufferId, &qr, &pending);
    if (err != err_NoError)
      dbg(DBG_USR2, "error,peek: %d", err);
    err = call QueryResultIntf.toTuplePtr(&qr, mQueryPtr, &t);
    if (err != err_NoError)
      dbg(DBG_USR2, "error,toTuplePtr: %d", err);
    pq = *(call DBBuffer.getSchema(mQueryPtr->bufferId));

    dbg(DBG_USR2,"num fields = %d\n", pq->numFields);
    err = call ParsedQueryIntf.getResultField(pq, &qr, 0, (char *)&result);
    if (err != err_NoError)
      dbg(DBG_USR2, "error, %d", err);
    else
      dbg(DBG_USR2,"tuple.val = %d\n",result); fflush(stdout);
  }

#endif


  /* ----------------------------------- DBBuffer Methods ------------------------------------- */

  /** Enqueue a result into the specified buffer 
     Note that if pending returns true, this routine may store a reference to r (so it had better not be on the stack!)
   */
  command TinyDBError DBBuffer.enqueue(uint8_t bufferId, QueryResultPtr r, bool *pending, ParsedQuery *pq) {
    Buffer *buf;
    uint16_t row;
    TinyDBError err = err_NoError;
    Handle *bufList;

    *pending = FALSE;

    switch (bufferId) {
    case kRADIO_BUFFER: {
      if (mState != kIDLE) return err_ResultBufferBusy;


      mCurResult = r;
      mCurQuery = pq;
      mCurResultIdx = 0;

      err = continueRadioBufferEnqueue(); //try to send the first result
      if (mState != kIDLE) *pending = TRUE;
      return err;
    }
    case kATTR_BUF:
    case kEVENT_BUF:
    case kCOMMAND_BUF:
    case kQUERY_BUF:
      return err_UnsupportedBuffer;
    default:
      err = getBuf(bufferId, &buf);
      if (err != err_NoError) return err;
      break;
    }

    switch (buf->type) {
    case kCOMMAND: 
      {
	//low byte of data should be command id!
	char *cmdStr = (char *)&buf->data;
	CommandDescPtr cmd = call CommandUse.getCommandById((uint8_t)(cmdStr[0]));


	if (cmd!=NULL) {
	  ParamVals params;
	  SchemaErrorNo schema_err;
	  char result[1];


	  switch (cmd->params.numParams) {
	  case 0:
	    params.numParams = 0;
	    break;