Реализация протокола RTS / CTS в Unetstack

Question

Я внедряю технику хэнсхейкинга RTS / CTS в UnetStack.Я написал код для реализации RTS / CTS между 2 узлами (для тестирования).Но я сталкиваюсь с некоторыми проблемами в этом.

Это в UnetStack IDE (версия 1.3).Ниже приведены проблемы, с которыми я сталкиваюсь.1. В fshrc.groovy я использовал API ReservationReq для отправки RTS.RxFrameNtf используется для получения уведомления о полученном сигнале, но когда я использую переменную 'rx', чтобы проверить, является ли принятый сигнал CTS, я получаю нулевое значение (строка 24 fshrc.groovy), которое указывает, что яне получает CTS.

Когда я отправляю ReservationReq на узел 2 с узла 1, я не уверен, какая функция в MySimpleHandshakeMac.groovy, processRequest или processMessage вызывается, поскольку ни один из его операторов печати не выполняется.

Код 1: fshrc.groovy

import org.arl.unet.*
import org.arl.unet.phy.*
import org.arl.unet.*
import org.arl.unet.phy.*
import org.arl.unet.mac.*
import org.arl.fjage.*

def mac = agentForService Services.MAC

subscribe phy

// add a closure to define the 'ping' command
send = { addr, value ->
  println "sending RTS to node $addr"
  phy << new ReservationReq(recipient: mac, to: addr, duration: 5.second)//not sure about the syntax
  println "RTS Sent"
  def msg = receive(RxFrameNtf, 1000)
  println "abc1"
  def rxNtf = receive({ it instanceof RxFrameNtf && it.from == addr}, 5000)
  println "abc2"
  println rxNtf
   def rx = pdu.decode(msg.data)
  if(rx.type == CTS_PDU && rxNtf && rxNtf.from == addr)
  {
    println "CTS Received at ${rxNtf.to} from ${rxNtf.from}}"
    phy << new DatagramReq(to: addr, protocol: Protocol.DATA, data: value)
  } 
}

Код 2: рукопожатие-sim.groovy

//! Simulation: 3-node network with ping daemons

import org.arl.fjage.Message
import org.arl.unet.*
import org.arl.mac.*
import org.arl.unet.phy.*

import org.arl.fjage.RealTimePlatform

platform = RealTimePlatform

// run simulation forever
simulate {
  node '1', address: 1, remote: 1101, location: [0, 0, 0], shell: true, stack: { container ->
    container.add 'hand', new MySimpleHandshakeMac()
    container.shell.addInitrc "${script.parent}/fshrc.groovy"
  }
  node '2', address: 2, remote: 1102, location: [1.km, 0, 0], shell:5102, stack: { container ->
    container.add 'hand', new MySimpleHandshakeMac()
  }
}

Код 3: MySimpleHandshakeMac.groovy

import org.arl.fjage.*
import org.arl.unet.*
import org.arl.unet.phy.*
import org.arl.unet.mac.*
import org.arl.unet.nodeinfo.*

class MySimpleHandshakeMac extends UnetAgent {

  ////// protocol constants

  private final static int PROTOCOL = Protocol.MAC

  private final static float  RTS_BACKOFF     = 2.seconds
  private final static float  CTS_TIMEOUT     = 5.seconds
  private final static float  BACKOFF_RANDOM  = 5.seconds
  private final static float  MAX_PROP_DELAY  = 2.seconds
  private final static int    MAX_RETRY       = 3
  private final static int    MAX_QUEUE_LEN   = 16

  ////// reservation request queue

  private Queue<ReservationReq> queue = new ArrayDeque<ReservationReq>()

  ////// PDU encoder/decoder

  private final static int RTS_PDU = 0x01
  private final static int CTS_PDU = 0x02

  private final static PDU pdu = PDU.withFormat {
    uint8('type')         // RTS_PDU/CTS_PDU
    uint16('duration')    // ms
  }

  ////// protocol FSM

  private enum State {
    IDLE, RTS, TX, RX, BACKOFF
  }

  private enum Event {
    RX_RTS, RX_CTS, SNOOP_RTS, SNOOP_CTS
  }

  private FSMBehavior fsm = FSMBuilder.build {

    int retryCount = 0
    float backoff = 0
    def rxInfo

    state(State.IDLE) {
      action {
        if (!queue.isEmpty()) {
          after(rnd(0, BACKOFF_RANDOM)) {
            setNextState(State.RTS)
          }
        }
        block()
      }
      onEvent(Event.RX_RTS) { info ->
        rxInfo = info
        setNextState(State.RX)
      }
      onEvent(Event.SNOOP_RTS) {
        backoff = RTS_BACKOFF
        setNextState(State.BACKOFF)
      }
      onEvent(Event.SNOOP_CTS) { info ->
        backoff = info.duration + 2*MAX_PROP_DELAY
        setNextState(State.BACKOFF)
      }
    }

    state(State.RTS) {
      onEnter {
        Message msg = queue.peek()
        def bytes = pdu.encode(type: RTS_PDU, duration: Math.ceil(msg.duration*1000))
        phy << new TxFrameReq(to: msg.to, type: Physical.CONTROL, protocol: PROTOCOL, data: bytes)
        after(CTS_TIMEOUT) {
          if (++retryCount >= MAX_RETRY) {
            sendReservationStatusNtf(queue.poll(), ReservationStatus.FAILURE)
            retryCount = 0
          }
          setNextState(State.IDLE)
        }
      }
      onEvent(Event.RX_CTS) {
        setNextState(State.TX)
      }
    }

    state(State.TX) {
      onEnter {
        ReservationReq msg = queue.poll()
        retryCount = 0
        sendReservationStatusNtf(msg, ReservationStatus.START)
        after(msg.duration) {
          sendReservationStatusNtf(msg, ReservationStatus.END)
          setNextState(State.IDLE)
        }
      }
    }

    state(State.RX) {
      onEnter {
        def bytes = pdu.encode(type: CTS_PDU, duration: Math.round(rxInfo.duration*1000))
        phy << new TxFrameReq(to: rxInfo.from, type: Physical.CONTROL, protocol: PROTOCOL, data: bytes)
        after(rxInfo.duration + 2*MAX_PROP_DELAY) {
          setNextState(State.IDLE)
        }
        rxInfo = null
      }
    }

    state(State.BACKOFF) {
      onEnter {
        after(backoff) {
          setNextState(State.IDLE)
        }
      }
      onEvent(Event.SNOOP_RTS) {
        backoff = RTS_BACKOFF
        reenterState()
      }
      onEvent(Event.SNOOP_CTS) { info ->
        backoff = info.duration + 2*MAX_PROP_DELAY
        reenterState()
      }
    }

  } // of FSMBuilder

  ////// agent startup sequence

  private AgentID phy
  private int addr

  @Override
  void setup() {
    register Services.MAC
  }

  @Override
  void startup() {
    phy = agentForService Services.PHYSICAL
    subscribe(phy)
    subscribe(topic(phy, Physical.SNOOP))
    add new OneShotBehavior({
      def nodeInfo = agentForService Services.NODE_INFO
      addr = get(nodeInfo, NodeInfoParam.address)
    })
    add(fsm)
  }

  ////// process MAC service requests

  @Override
  Message processRequest(Message msg) {
    println "processRequest"
    switch (msg) {
      case ReservationReq:
        if (msg.to == Address.BROADCAST || msg.to == addr) return new Message(msg, Performative.REFUSE)
        if (msg.duration <= 0 || msg.duration > maxReservationDuration) return new Message(msg, Performative.REFUSE)
        if (queue.size() >= MAX_QUEUE_LEN) return new Message(msg, Performative.REFUSE)
        queue.add(msg)
        fsm.restart()      // tell fsm to check queue, as it may block if the queue is empty
        println "Request Accepted"
        return new ReservationRsp(msg)
      case ReservationCancelReq:
      case ReservationAcceptReq:
      case TxAckReq:
        return new Message(msg, Performative.REFUSE)
    }
    return null
  }

  ////// handle incoming MAC packets

  @Override
  void processMessage(Message msg) {
    println "processMessage"
    if (msg instanceof RxFrameNtf && msg.protocol == PROTOCOL) {
      def rx = pdu.decode(msg.data)
      def info = [from: msg.from, to: msg.to, duration: rx.duration/1000.0]
      if (rx.type == RTS_PDU) fsm.trigger(info.to == addr ? Event.RX_RTS : Event.SNOOP_RTS, info)
      else if (rx.type == CTS_PDU) fsm.trigger(info.to == addr ? Event.RX_CTS : Event.SNOOP_CTS, info)
    }
  }


  ////// expose parameters that are expected of a MAC service

  final int reservationPayloadSize = 0            // read-only parameters
  final int ackPayloadSize = 0
  final float maxReservationDuration = 65.535

  @Override
  List<Parameter> getParameterList() {            // publish list of all exposed parameters
    return allOf(MacParam)
  }

  boolean getChannelBusy() {                      // channel is considered busy if fsm is not IDLE
    return fsm.currentState.name != State.IDLE
  }

  float getRecommendedReservationDuration() {     // recommended reservation duration is one DATA packet
    return get(phy, Physical.DATA, PhysicalChannelParam.frameDuration)
  }

  ////// utility methods

  private void sendReservationStatusNtf(ReservationReq msg, ReservationStatus status) {
    send new ReservationStatusNtf(recipient: msg.sender, requestID: msg.msgID, to: msg.to, from: addr, status: status)
  }
}

IЯ печатаю RxNtf.«null» печатается как значение RxNtf.

Answer 1

Когда вы отправляете ReservationReq, агент MAC отправит RTS и получит CTS.Хотя вы можете иметь возможность подписаться на PHY напрямую для прослушивания CTS, это не будет правильным подходом для использования MAC.Вам следует дождаться MAC ReservationStatusNtf с ReservationStatus.START, чтобы начать передачу.

Вот несколько примеров кода, чтобы показать, как работает подход:

def mac = agentForService Services.MAC
if (mac) {
  // send reservation request
  def req = new ReservationReq(recipient: mac, to: addr, duration: 5.second)
  def rsp = request req
  if (rsp && rsp.performative == Performative.AGREE) {
    // wait for a channel reservation notification
    def ntf = receive(ReservationStatusNtf, timeout)
    if (ntf && ntf.requestID == req.msgID && ntf.status == ReservationStatus.START) {
      // request granted, make your transmission
      phy << new DatagramReq(to: addr, protocol: Protocol.DATA, data: value)
    }
  }
}

Кроме того факта, что непосредственное прослушивание пакета CTS будет хрупким (так как MySimpleHandshakeMac может изменить детали протокола), причина, по которой ваш подход терпит неудачу, заключается в том, что в MySimpleHandshakeMac имеется случайный откат (BACKOFF_RANDOM, установленный на 5 секунд), в то время как время ожидания в fshrc составляло всего 1 секунду.Таким образом, вы, скорее всего, вышли на тайм-аут даже до того, как отправили RTS, не говоря уже о получении CTS.