dbbufferc.nc

用于传感器网络的节点操作系统 TinyOS 结构设计非常有意思

    uint16_t row = buf->nextFull;
    Handle *bufList;

    getBuf(bufferId, &buf);
    bufList = (Handle *)(*buf->u.bufdata);
    data = bufList[row];
    if (data != NULL) {
      call MemAlloc.free(data);
      bufList[row] = NULL;
    }
    buf->nextFull++;
    if (buf->nextFull >= buf->numRows) buf->nextFull = 0;    
  }
    
  /** Copy the top most result in the buffer into buf
     @return err_NoMoreResults if no results are available
  */
  command TinyDBError DBBuffer.peek(uint8_t bufferId, QueryResult *qr, bool *pending) {
    Buffer *buf;
    TinyDBError err = getBuf(bufferId, &buf);
    uint16_t row;
    Handle *bufList;
    
    //bool prev = call Interrupt.disable();    
    if (mState != kIDLE) {
      //if (prev) call Interrupt.enable();
      return err_ResultBufferBusy;
    }
    mState = kREAD_ROW;
    //if (prev) call Interrupt.enable();

    *pending = FALSE;
    
    if (err != err_NoError) goto done;

    row= buf->nextFull;
    if (row == buf->nextFree && buf->numRows != 1) { //special case single row...
      err =  err_NoMoreResults;
      goto done;
    }

    switch(buf->type) {
    case kRAM:
      bufList = (Handle *)(*buf->u.bufdata);
      if (bufList[row] == NULL) {
	err =  err_NoMoreResults; 
	goto done;
      }
      call QueryResultIntf.fromBytes((char *)*bufList[row], qr, *buf->qh,0);
      //memcpy(qr, &(*buf->u.bufdata)[row*buf->rowSize],buf->rowSize);
      break;
    default:
      err = err_UnknownError;
    }

  done:
    mState = kIDLE;
    return err;
  }
  
  /** Copy the nth result in the buffer into buf
     @return err_NoMoreResults  if idx > getSize() or if buffer[idx] is empty (unset)
  */
  command TinyDBError DBBuffer.getResult(uint8_t bufferId, uint16_t idx, QueryResult *qr, bool *pending) {
    Buffer *buf;
    TinyDBError err = getBuf(bufferId, &buf);
    uint16_t size;
    uint16_t row;
    Handle *bufList;

    //bool prev = call Interrupt.disable();    
    if (mState != kIDLE) {
      //if (prev) call Interrupt.enable();
      return err_ResultBufferBusy;
    }
    mState = kREAD_ROW;
    //if (prev) call Interrupt.enable();

    *pending = FALSE;
    
    if (err != err_NoError) goto done;
    
    if (buf->nextFull == buf->nextFree && buf->numRows != 1) {
      err =  err_NoMoreResults;
      goto done;
    }
    call DBBuffer.size(bufferId, &size);
    if (idx >= size) {
      err = err_NoMoreResults;
      goto done;
    }
    
    //see comment at beginning of document for info on slot management & indexing
    if (buf->nextFull > buf->nextFree) {
      if (idx >= (buf->numRows - buf->nextFull)) {
	row = idx - (buf->numRows - buf->nextFull);
      } else
	row = buf->nextFull + idx;
    } else {
      row = buf->nextFull + idx;
    }

    bufList = (Handle *)(*buf->u.bufdata);
    call QueryResultIntf.fromBytes((char *)*bufList[row], qr, *buf->qh,0);
    //memcpy(qr, &(*buf->u.bufdata)[row*buf->rowSize], buf->rowSize);
    
  done:
    mState = kIDLE;
    return err;
  }

  /* Return the number of used rows in the buffer */
    command TinyDBError DBBuffer.size(uint8_t bufferId, uint16_t *size) {
      Buffer *buf;
      TinyDBError err = getBuf(bufferId, &buf);
      
      if (err != err_NoError) return err;
      
    //see comment at beginning of document for info on slot management & indexing
      if (buf->nextFull == buf->nextFree)
	*size = 0;
      else if (buf->nextFull > buf->nextFree) {
	*size = (buf->numRows - buf->nextFull) + buf->nextFree;
      } else {
	*size = buf->nextFree - buf->nextFull;
      }
      return err_NoError;
    }
  
    /** Allocate the buffer with the specified size 
	sizes is an array of sizes of each field, with one entry per field
	<p>
	Note that this may keep a reference to schema until after pending is complete.
	<p>
	Signals allocComplete when allocation is complete if *pending is true on return
	@param bufferId The buffer to allocate
	@param type The type of the buffer (see DBBuffer.h -- only kRAM and kRADIO are supported)
	@param size The size (in rows) of the buffer
	@param policy The eviction policy to use with the buffer (see DBBuffer.h)
	@param schema The schema (layout) of rows in this buffer (expressed as a query)
	@param pending On return, set to true if the buffer is still being allocated (expect allocComplete if
	               true).
	@param data is currently unused
	@return err_UnsupportedPolicy if the specified policy can't be applied
    */
  command TinyDBError DBBuffer.alloc(uint8_t bufferId, BufferType type, uint16_t size, BufferPolicy policy,
				     ParsedQuery *schema, bool *pending, long data) {
    
    Buffer *buf;
    

    *pending = FALSE;
    
    if (mState != kIDLE) return err_ResultBufferBusy;

    if (bufferId >= kNUM_BUFS)
      return err_InvalidIndex;
    
    if (type == kRADIO || type == kEEPROM)
      return err_UnsupportedBuffer; // can't do these right now

    if (mUsedBufs & (1 << bufferId))
      return err_ResultBufferInUse;
    
    mState = kALLOC_SCHEMA;

    buf = &mBuffers[bufferId];


    buf->type = type;
    buf->policy = policy;
    buf->numRows = size;
    buf->nextFree = 0;
    buf->nextFull = 0;
    buf->data = data;

    switch (type) {
    case kRAM:

      buf->rowSize = sizeof(Handle); //rows are handles into memory
      buf->len = buf->rowSize * buf->numRows;

      mCurBuffer = bufferId;
      mAllocSchema = schema;
      
      //now, we have to allocate buf->u.bufdata and buf->qh
      if (call MemAlloc.allocate((HandlePtr)(&buf->qh), call ParsedQueryIntf.pqSize(schema)) == SUCCESS) {
	*pending = TRUE;
	return err_NoError;
      } else {
	mState = kIDLE;
	return err_OutOfMemory;
      }
      break;
    case kCOMMAND:
      mState = kIDLE;
      mUsedBufs |= (1 << bufferId); 
      return err_NoError;
      break;
    default: 
      break;
    }


    return err_NoError;


  }
  
  /** @return the number of rows in this buffer */
  command uint16_t DBBuffer.maxSize(uint8_t bufferId) {
    Buffer *buf;
    TinyDBError err = getBuf(bufferId, &buf);
    if (err != err_NoError) return 0;
    
    return buf->numRows;
  }
  
  /** @return the schema of the result */
  command ParsedQuery **DBBuffer.getSchema(uint8_t bufferId) {
    Buffer *buf;
    TinyDBError err = getBuf(bufferId, &buf);
    if (err != err_NoError) return NULL;
    return buf->qh;
  }

  /* Return the next unused buffer id, or err_OutOfMemory, 
     if no more buffers are available */
  command TinyDBError DBBuffer.nextUnusedBuffer(uint8_t *bufferId) {
    int i;

    for (i = 0; i < kNUM_BUFS; i++) {
      if (!(mUsedBufs & (1<<i))) { 
	*bufferId = i;
	return err_NoError;
      }
    }
    
    return err_OutOfMemory;

  }
  
  /** @return the buffer id that corresponds to the specified query id
     in bufferId, or err_InvalidIndex if no such buffer exists 
  */
  command TinyDBError DBBuffer.qidToBuffer(uint8_t qid, uint8_t *bufferId) {
    int i;

    //don't look at the the radio buffer!
    for (i = 1; i < kNUM_BUFS; i++) {
      if ((mUsedBufs & (1<<i))) {
	if ((**mBuffers[i].qh).qid == qid) {
	  *bufferId = i;
	  return err_NoError;
	}
      }
    }

    return err_InvalidIndex;
  }



  /* --------------------- Default event handlers ------------------------- */
  default event result_t DBBuffer.resultReady(uint8_t bufferId) {
    return SUCCESS;
  }

  default event result_t DBBuffer.getNext(uint8_t bufferId) {
    return SUCCESS;
  }

  default event result_t DBBuffer.allocComplete(uint8_t bufferId, TinyDBError result) {

    return SUCCESS;
  }

  default event result_t DBBuffer.putComplete(uint8_t bufferId, QueryResult *buf, TinyDBError result) {

    return SUCCESS;
  }

  /* --------------------- Event handlers ------------------------- */

  event result_t MemAlloc.allocComplete(HandlePtr handle, result_t success) {
    Handle *bufList;

	//Handle h = bufList[buf->nextFull];
      if (mState == kIDLE || mState == kREAD_ROW) //not for us
        return SUCCESS; 

      if (success) {
        switch (mState) { 
        case kALLOC_SCHEMA:
	  //need to copy schema over!
	  memcpy (**handle,(char *)mAllocSchema,call ParsedQueryIntf.pqSize(mAllocSchema));
	  //now we have to alloc the memory buffer */
	  mState = kALLOC_BUFFER; 
	  
	  if (call MemAlloc.allocate((HandlePtr)&mBuffers[mCurBuffer].u.bufdata, mBuffers[mCurBuffer].len) == FAIL) { 
	    mState = kIDLE; 
	    signal DBBuffer.allocComplete(mCurBuffer, err_UnknownError); 
	  } 
	  break; 
        case kALLOC_BUFFER: 
	  mUsedBufs |= (1 << mCurBuffer); 

	  memset(**handle, 0, mBuffers[mCurBuffer].len); //clear to 0
	  mState = kIDLE; 
#ifdef kDEBUG
	  enqueueResult();
#endif
	  signal DBBuffer.allocComplete(mCurBuffer, err_NoError); 
	  break; 
	case kALLOC_ROW: {

	  Buffer *buf = &mBuffers[mCurBuffer];
	  bufList = (Handle *)(*buf->u.bufdata);
	  bufList[mCurRow] = *handle;
	  call MemAlloc.lock(*handle);
	  call QueryResultIntf.toBytes(mCurResult, *buf->qh, (char *)**handle);
	  call MemAlloc.unlock(*handle);
	  mState = kIDLE;

#ifdef kDEBUG
	  readResult();
	  enqueueResult();
#endif
	  signal DBBuffer.putComplete(mCurBuffer, mCurResult, err_NoError);
	  break;
	}
	case kREAD_ROW:
	case kIDLE: //checked for this above -- just do this to stop compiler warnings
	  break;
        } 
      } else { 
        mState = kIDLE; 
	switch (mState) {
	case kALLOC_SCHEMA:
	case kALLOC_BUFFER:
	  signal DBBuffer.allocComplete(mCurBuffer, err_UnknownError); 
	  break;
	case kALLOC_ROW:
	  signal DBBuffer.putComplete(mCurBuffer, mCurResult, err_UnknownError);
	case kIDLE: //checked for this above -- just do this to stop compiler warnings
	case kREAD_ROW:
	  break; 
	}
      } 
    
    return SUCCESS;
  }

  event result_t MemAlloc.compactComplete() {
    return SUCCESS;
  }

  event result_t MemAlloc.reallocComplete(Handle handle, result_t success) {
    if (mState != kIDLE && handle == mCurRowHandle)
      return signal MemAlloc.allocComplete(&mCurRowHandle,success);
    else
      return SUCCESS; //not for us
    //return SUCCESS;
  }

  event result_t QueryProcessor.queryComplete(ParsedQueryPtr q) {
      uint8_t bufid; 

      if (call DBBuffer.qidToBuffer(q->qid, &bufid) == err_NoError) { 
        Buffer *buf = &mBuffers[bufid]; 

        //delete this buffer 
        call MemAlloc.free((Handle)buf->qh); 
        if (buf->type == kRAM) { 
	  call MemAlloc.free((Handle)buf->u.bufdata); 
        } 
        mUsedBufs &= (0xFF | (1 << bufid)); //mark this buffer as unused 
      } 

    return SUCCESS;
  }

  event result_t CommandUse.commandDone(char *commandName, char *resultBuf, SchemaErrorNo err) {
    return SUCCESS;
  }
  
  /* --------------------- Private Routines ------------------------- */

  /** (PRIVATE) Adjust the next free row in the buffer
     @return err_OutOfMemory if the policy indicates that no more inserts should be allowed 
  */

  TinyDBError calcNextFreeRow(Buffer *buf) {
    switch (buf->policy) {
    case EvictOldestPolicy:
      //just a circular buffer

      buf->nextFree++;
      if (buf->nextFree >= buf->numRows) 
	buf->nextFree = 0;

      //if current free reaches head of queue, advance head of queue (which is pointing at oldest element)
      //also deallocate current nextFull item
      if (buf->nextFree == buf->nextFull) {

	//we don't actually dispose of the memory anymore -- instead, we realloc, which can be a lot faster
	/* Handle *bufList = (Handle *)(*buf->u.bufdata);
	   Handle h = bufList[buf->nextFull];
	
	   if (h != NULL) call MemAlloc.free(h);
	   bufList[buf->nextFull] = NULL;
	*/

	buf->nextFull++;
	if (buf->nextFull >= buf->numRows)
	  buf->nextFull = 0;
      }


      break;
    }
    return err_NoError;

  }

  /** (PRIVATE) Return a pointer to buffer bufId in buf if bufId is a valid buf.
     Otherwise, return err_InvalidIndex
  */
  TinyDBError getBuf(uint8_t bufId, Buffer **buf) {
    if (bufId >= kNUM_BUFS)
      return err_IndexOutOfBounds; //unknown schema
    if ((mUsedBufs & (1 << bufId)) == 0)
      return err_InvalidIndex;
    *buf = &mBuffers[bufId];
    return err_NoError;
  }
  

}