delugepagetransferm.nc

来自「tinyos最新版」· NC 代码 · 共 661 行 · 第 1/2 页

// $Id: DelugePageTransferM.nc,v 1.28 2004/08/28 19:50:59 jwhui Exp $

/*									tab:4
 *
 *
 * "Copyright (c) 2000-2004 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."
 *
 */

/**
 * Handles the transfer of individual data pages between neighboring
 * nodes.
 *
 * @author Jonathan Hui <[e-mail]>
 */

module DelugePageTransferM {
  provides {
    interface DelugePageTransfer as PageTransfer;
    interface StdControl;
  }
  uses {
    interface BitVecUtils;
    interface DelugeImgStableStore as StableStore;
    interface DelugeMetadata as Metadata;
    interface Random;
    interface ReceiveMsg as ReceiveDataMsg;
    interface ReceiveMsg as ReceiveReqMsg;
    interface Leds;
    interface SendMsg as SendDataMsg;
    interface SendMsg as SendReqMsg;
    interface SharedMsgBuf;
    interface StdControl as SubControl;
    interface Timer as ReqTimer;
#ifdef DELUGE_DBG
    interface DelugeDbg;
#endif
  }
}

implementation {

  typedef struct {
    uint16_t dataSourceAddr;
    uint8_t  numReqTriesLeft;
    bool     suppressReq;
  } RxState;

  typedef struct {
    uint8_t imgToSend;
    uint8_t pgToSend;
  } TxState;

  typedef struct {
    union {
      RxState rx; 
      TxState tx; 
    };
  } RxTxState;

  typedef struct {
    uint8_t pktNum;
    uint8_t data[DELUGE_PKT_PAYLOAD_SIZE];
  } DataQEntry;

  // Data Write Queue
  DataQEntry dataQ[DELUGE_DATA_Q_SIZE];
  uint8_t  dataQHead, dataQLen;

  // send/receive page buffers, and state variables for buffers
  uint8_t  pktsToSend[DELUGE_PKT_BITVEC_SIZE];    // bit vec of packets to send
  uint8_t  pktsToReceive[DELUGE_PKT_BITVEC_SIZE]; // bit vec of packets to receive

  // state variables
  uint8_t   state;
  RxTxState vars;

  enum {
    S_DISABLED,
    S_IDLE,     
    S_SENDING,
    S_TX_LOCKING,
    S_RECEIVING,
    S_RX_LOCKING,
  };

  void     changeState(uint8_t _state);
  void     setupDataMsg();

  command result_t StdControl.init() {

    result_t result = call SubControl.init();

    // initialize state variables
    changeState(S_DISABLED);

    return result;

  }

  command result_t StdControl.start() {

    result_t result = call SubControl.start();

    // get ready to receive next page
    dataQHead = dataQLen = 0;
    memset(pktsToReceive, 0xff, DELUGE_PKT_BITVEC_SIZE);

    // enable receives and sends
    changeState(S_IDLE);

    return result;

  }

  command result_t StdControl.stop() {
    // disable receives and sends
    changeState(S_DISABLED);
    return SUCCESS;
  }

  command result_t PageTransfer.setNewSource(uint16_t sourceAddr) {

    if (call Metadata.getNextIncompleteImage() == DELUGE_INVALID_IMGNUM)
      return FAIL;

    if (state == S_IDLE) {
      // currently idle, so request data from source
      changeState(S_RX_LOCKING);
      vars.rx.dataSourceAddr = sourceAddr;
      vars.rx.suppressReq = FALSE;

      // randomize request to prevent collision
      call ReqTimer.start(TIMER_ONE_SHOT, DELUGE_MIN_DELAY + 
			  call Random.rand() % DELUGE_MAX_REQ_DELAY);
    }

    return SUCCESS;

  }

  task void retrySetupDataMsg() {
    setupDataMsg();
  }

  uint32_t calcOffset(pgnum_t pgNum, uint8_t pktNum) {
    return (uint32_t)pgNum*(uint32_t)DELUGE_BYTES_PER_PAGE
      + (uint32_t)pktNum*(uint32_t)DELUGE_PKT_PAYLOAD_SIZE;
  }

  void setupDataMsg() {
    TOS_MsgPtr pMsgBuf = call SharedMsgBuf.getMsgBuf();
    DelugeDataMsg* pDataMsg = (DelugeDataMsg*)(pMsgBuf->data);
    uint32_t offset;
    uint16_t length;
    uint16_t nextPkt;
    result_t result;

    if (state != S_SENDING && state != S_TX_LOCKING)
      return;

    if (state == S_TX_LOCKING) {
      if (call SharedMsgBuf.lock() == FAIL) {
	post retrySetupDataMsg();
	return;
      }
      changeState(S_SENDING);
      pDataMsg->vNum = call Metadata.getVNum(vars.tx.imgToSend);
      pDataMsg->imgNum = vars.tx.imgToSend;
      pDataMsg->pgNum = vars.tx.pgToSend;
      pDataMsg->pktNum = 0;
    }
    
    if (pDataMsg->pktNum < DELUGE_PKTS_PER_PAGE) {
      if (!call BitVecUtils.indexOf(&nextPkt, pDataMsg->pktNum, pktsToSend, 
				    DELUGE_PKTS_PER_PAGE)
	  && !call BitVecUtils.indexOf(&nextPkt, 0, pktsToSend, 
				       DELUGE_PKTS_PER_PAGE)) {
	// no more packets to send
	dbg(DBG_USR1, "DELUGE: SEND_DONE\n");
	changeState(S_IDLE);
	return;
      }
      pDataMsg->pktNum = (uint8_t)nextPkt;

      length = DELUGE_PKT_PAYLOAD_SIZE;
      if (pDataMsg->pktNum == DELUGE_PKTS_PER_PAGE - 1)
	length -= 2;
      
      offset = calcOffset(pDataMsg->pgNum, pDataMsg->pktNum);      
      result = call StableStore.getImgData(pDataMsg->imgNum, offset, pDataMsg->data, length);
    }
    else {
      result = call StableStore.getPageCrc(pDataMsg->imgNum, pDataMsg->pgNum, 
					   &(pDataMsg->data[DELUGE_PKT_PAYLOAD_SIZE-2]));
    }

    if (result == FAIL)
      post retrySetupDataMsg();

  }

  event result_t ReqTimer.fired() {

    TOS_MsgPtr pMsgBuf = call SharedMsgBuf.getMsgBuf();
    DelugeReqMsg* pReqMsg = (DelugeReqMsg*)(pMsgBuf->data);

    if (state != S_RECEIVING && state != S_RX_LOCKING)
      return SUCCESS;

    if (state == S_RX_LOCKING) {
      if (call SharedMsgBuf.lock() == FAIL) {
	call ReqTimer.start(TIMER_ONE_SHOT, DELUGE_FAILED_SEND_DELAY);	
	return SUCCESS;
      }
      // update address of source node
      changeState(S_RECEIVING);
      pReqMsg->dest = vars.rx.dataSourceAddr;
      vars.rx.numReqTriesLeft = DELUGE_MAX_NUM_REQ_TRIES;
      pReqMsg->imgNum = call Metadata.getNextIncompleteImage();
      pReqMsg->pgNum = call Metadata.getNextIncompletePage();
      pReqMsg->vNum = call Metadata.getVNum(pReqMsg->imgNum);
      memcpy(pReqMsg->requestedPkts, pktsToReceive, DELUGE_PKT_BITVEC_SIZE);
    }

    if (vars.rx.suppressReq) {
      // suppress request
      call ReqTimer.start(TIMER_ONE_SHOT, DELUGE_NACK_TIMEOUT +
			  call Random.rand() % DELUGE_NACK_TIMEOUT);
      vars.rx.suppressReq = FALSE;
      return SUCCESS;
    }

    if (vars.rx.numReqTriesLeft == 0) {
      // tried too many times, give up
      changeState(S_IDLE);
      return SUCCESS;
    }

    {
      uint16_t dest = (pReqMsg->dest == TOS_UART_ADDR) ? TOS_UART_ADDR : TOS_BCAST_ADDR;
      if (call SendReqMsg.send(dest, sizeof(DelugeReqMsg), pMsgBuf) == SUCCESS) {
	vars.rx.numReqTriesLeft--;
      }
      else {
	// send failed, wait a bit and try again
	call ReqTimer.start(TIMER_ONE_SHOT, DELUGE_FAILED_SEND_DELAY);
      }
    }
    return SUCCESS;
  }

  event TOS_MsgPtr ReceiveReqMsg.receive(TOS_MsgPtr pMsg) {

    DelugeReqMsg *rxReqMsg = (DelugeReqMsg*)(pMsg->data);
    uint8_t imgNum, tmpImgNum;
    pgnum_t pgNum, tmpPgNum;
    int i;

#ifdef DELUGE_DBG
    call DelugeDbg.rxReqMsg();
#endif

    if (state == S_DISABLED)
      return pMsg;

    imgNum = call Metadata.getNextIncompleteImage();
    pgNum = call Metadata.getNextIncompletePage();

    if (state == S_SENDING || state == S_TX_LOCKING) {
      tmpImgNum = vars.tx.imgToSend;
      tmpPgNum = vars.tx.pgToSend;
    }
    else {
      tmpImgNum = imgNum;
      tmpPgNum = pgNum;
    }

    if (rxReqMsg->imgNum < tmpImgNum
	|| (rxReqMsg->imgNum == tmpImgNum
	    && (rxReqMsg->vNum - call Metadata.getVNum(tmpImgNum) > 0
		|| (rxReqMsg->vNum == call Metadata.getVNum(tmpImgNum)
		    && rxReqMsg->pgNum < tmpPgNum)))) {
      if (state == S_IDLE)
	signal PageTransfer.overheardOlderData();
      else if (state == S_SENDING || state == S_TX_LOCKING) {
	changeState(S_IDLE);
	memset(pktsToSend, 0x0, DELUGE_PKT_BITVEC_SIZE);
      }
    }

    if (rxReqMsg->dest != TOS_LOCAL_ADDRESS) {
      if ((state == S_RECEIVING || state == S_RX_LOCKING)
	  && rxReqMsg->imgNum == imgNum
	  && rxReqMsg->vNum == call Metadata.getVNum(imgNum)
	  && rxReqMsg->pgNum <= pgNum) {
	// suppress next request since similar request has been overheard
	vars.rx.suppressReq = TRUE;
      }
      return pMsg;
    }

    if (rxReqMsg->pgNum >= call Metadata.getNumPgsComplete(rxReqMsg->imgNum)
	|| rxReqMsg->vNum != call Metadata.getVNum(rxReqMsg->imgNum)) {
      // don't have this page, ignore request
      return pMsg;
    }

    if (state == S_IDLE
	|| ((state == S_SENDING || state == S_TX_LOCKING)
	    && rxReqMsg->imgNum == vars.tx.imgToSend
	    && rxReqMsg->vNum == call Metadata.getVNum(vars.tx.imgToSend)
	    && rxReqMsg->pgNum == vars.tx.pgToSend)) {
      // take union of packet bit vectors