puma.cc

组播路由协议

                    struct ACKTimeoutMessage* message = 
                                            (ACKTimeoutMessage*)(event->data_);
                    MulticastGroup* group = 
                              multicastGroups_.find(message->multicastAddress_);
                    fightElection(group);
                    break;
                }
                case PARTITION_ACK_TIMEOUT: {
                    struct ACKTimeoutMessage* message = 
                                            (ACKTimeoutMessage*)(event->data_);
                    MulticastGroup* group = 
                              multicastGroups_.find(message->multicastAddress_);
                    fightMultipleGroupElectionAfterPartition(group);
                    break;
                }
                case SEND_PENDING_MA: {
                    nsaddr_t* multicastAddress = (nsaddr_t*)(event->data_);
                    MulticastGroup* group = 
                                       multicastGroups_.find(*multicastAddress);
                    sendASingleMA(group);
                    group->delayedMAPending() = false;
                    break;
                }
                case DATA_ACK_TIMEOUT: {
                    struct ACKTimeoutMessage* message = 
                                            (ACKTimeoutMessage*)(event->data_);
                    MulticastGroup* group = 
                              multicastGroups_.find(message->multicastAddress_);
                    ConnectivityListItem* item = new ConnectivityListItem(
                               message->nextHop_, 0, 0, INVALID_ADDRESS, false);
                    if (!(wasACKReceived(message->nextHop_)))
                        group->connectivityList_->remove(item);
                    break;
                }
                case SEND_NEXT_MA: {
                    nsaddr_t* multicastAddress = (nsaddr_t*)(event->data_);
                    MulticastGroup* group = 
                                       multicastGroups_.find(*multicastAddress);
                    if (shouldISendMA(group))
                        sendNextMA(group);
                    break;
                }
            }
        }
}

bool
PUMA::wasACKReceived(nsaddr_t destination) {
        int i, j;
        ackTable_.removeObsoleteEntries();
        i = ackTable_.getFirstOccupiedSlot();
        for (j = 0; j < ackTable_.numberOfACKTableEntries(); j++) {
            if ((ackTable_.elementAt(i)).heardFrom() == destination)
                return true;
            i++;
            i%=NUMBER_OF_ACK_TABLE_ENTRIES;
        }
        return false;
}

bool
PUMA::shouldISendMA(MulticastGroup* group) {
       if (group->lastMAOriginated() == NEVER)
           return true;
       if (floor(((NOW - group->lastMAOriginated())*1000+0.5)/1000) >=
MULTICAST_ANNOUNCEMENT )
           return true;
       return false;
}

void
PUMA::acceptDataPacket(Packet* p) {
        // this copy is because the transport free the packet up
        dmux_->recv(p->copy(), (Handler*)0);
}

void
PUMA::handleProtocolPacket(Packet* p) {
        struct hdr_puma *ph = HDR_PUMA(p);
        if (ph->MACount() > 0)
            handleBucket(p);
        else
            handleDataPacketFromNetwork(p);
};

void
PUMA::handleDataPacketFromNetwork(Packet* p) {
        struct hdr_cmn   *ch = HDR_CMN(p);
        struct hdr_ip    *ih = HDR_IP(p);
        MulticastGroup* group = multicastGroups_.find(ih->daddr());
        recordACK(ch->prev_hop_);
        MessageCacheItem* temp = new MessageCacheItem(ih->saddr(), ch->uid());
        if (group != NULL && ih->saddr() != id_  &&
            !(messageCache_.find(temp)) && ih->ttl() > 0) {
            if (amIMeshMember(group)) {
                ch->prev_hop_ = id_;
                processMulticastPacket(p);
            }
            else if (doIForwardThisPacket(group, p)) {
                ch->prev_hop_ = id_;
                forwardFreshDataPacket(group, p);
                setRetransmissionTimer(
                group->connectivityList_->start_->nextHop(), DATA_ACK_TIMEOUT, 
                               ACK_TIMEOUT_INTERVAL, group->multicastAddress());
            } else
                Packet::free(p);
        }
        else
            Packet::free(p);
}

void
PUMA::processMulticastPacket(Packet* p) {
        struct hdr_cmn   *ch = HDR_CMN(p);
        struct hdr_ip    *ih = HDR_IP(p);
        MulticastGroup* group = multicastGroups_.find(ih->daddr());
        (ih->ttl())--;
        if (group->receiver())
            acceptDataPacket(p);
        MessageCacheItem* temp = new MessageCacheItem(ih->saddr(), ch->uid());
        messageCache_.append(temp);
        ch->direction() = hdr_cmn::DOWN;
        ch->addr_type() = NS_AF_INET;
        ch->next_hop()  = IP_BROADCAST;
        Scheduler::instance().schedule(target_, p,
                                       BROADCAST_JITTER * random_.uniform(1.0));
}

void
PUMA::fightMultipleGroupElectionAfterPartition(MulticastGroup* group) {
        if (!(group->partitionConfirmationWait())) {
            group->partitionConfirmationWait() = true;
            sendAPartitionConfirmationRequest(group);
            setRetransmissionTimer(INVALID_ADDRESS, PARTITION_ACK_TIMEOUT, 
                     PARTITION_ACK_TIMEOUT_INTERVAL, group->multicastAddress());
        }
        else {
            group->partitionConfirmationWait() = false;
            if (isPartitionFlagSet(group) && group->receiver())
                becomeCoreAfterPartition(group);
	}
}

void
PUMA::fightElection(MulticastGroup* group) {
        if (!(group->neighborConfirmationWait())) {
            group->neighborConfirmationWait() = true;
            sendASingleMA(group);
            setRetransmissionTimer(INVALID_ADDRESS, JOIN_ACK_TIMEOUT, 
                               ACK_TIMEOUT_INTERVAL, group->multicastAddress());
        }
        else {
            group->neighborConfirmationWait() = false;
            if (group->coreId() == INVALID_ADDRESS) {
                group->receiver() = true;
                becomeCore(group);
            }
            else {
                becomeReceiver(group);
            }
        }
}

void
PUMA::handleJoinFromTransport(nsaddr_t groupAddress) {
        MulticastGroup* group = multicastGroups_.find(groupAddress);
        if (group == NULL) {
            group = new MulticastGroup(groupAddress, INVALID_ADDRESS);
            multicastGroups_.append(group);
            fightElection(group);
        } else if (group->coreId() == INVALID_ADDRESS)
            fightElection(group);
        else
            becomeReceiver(group);
}

void
PUMA::becomeReceiver(MulticastGroup* group) {
        if (!(group->receiver())) {
            group->receiver() = true;
            if (!(group->meshMember()))
                sendASingleMA(group);
        }
}

bool
PUMA::amIMeshMember(MulticastGroup* group) {
        if (group->receiver())
            return true;
        else if (amILinkNode(group))
            return true;
        else
            return false;
}

bool
PUMA::amILinkNode(MulticastGroup* group) {
        if (group == NULL)
            return false;
        else {
            if (getNumberOfMeshChildren(group->connectivityList_->start_) == 0)
                return false;
            else
                return true;
        }
}

bool
PUMA::isPartitionFlagSet(MulticastGroup* group) {
        if (group->lastMAReceived() == NEVER || 
            ((NOW  - group->lastMAReceived()) < 
             (3  * MULTICAST_ANNOUNCEMENT)))
            return false;
        else
            return true;
}

void
PUMA::updateGroup(MulticastGroup* group, MulticastAnnouncement ma, 
                       nsaddr_t sourceAddress) {
        group->reset();
        group->lastSequenceNumber() = ma.sequenceNumber_;
        group->coreId() = ma.coreId_;
        group->lastMAReceived() = NOW;
        ConnectivityListItem* temp = 
            new ConnectivityListItem(sourceAddress, ma.distanceToCore_ + 1, 
            ma.sequenceNumber_, ma.nextHop_, ma.meshMember_);
        group->connectivityList_->addSorted(temp);
}

void
PUMA::handleLeaveFromTransport(nsaddr_t groupAddress) {
        MulticastGroup* group = multicastGroups_.find(groupAddress);
        if (group != NULL && group->receiver()) {
            group->receiver() = false;
            if (group->coreId() == id_)
                group->reset();
        }
}

void
PUMA::becomeCore(MulticastGroup* group) {
        group->coreId() = id_;
        sendNextMA(group);
}

void
PUMA::becomeCoreAfterPartition(MulticastGroup* group) {
        group->reset();
        group->coreId() = id_;
        sendNextMA(group);
}

void
PUMA::forwardFreshDataPacket(MulticastGroup* group, Packet* p) {
        struct hdr_cmn   *ch = HDR_CMN(p);
        struct hdr_ip    *ih = HDR_IP(p);
        if (group->connectivityList_->start_ == NULL) {
            sendASingleMA(group);

        }
        else {
            ch->direction() = hdr_cmn::DOWN;
            ch->next_hop()  = IP_BROADCAST;
            (ih->ttl())--;
            Scheduler::instance().schedule(target_, p,
                                       BROADCAST_JITTER * random_.uniform(1.0));
            MessageCacheItem* temp= new MessageCacheItem(ih->saddr(),ch->uid());
            messageCache_.append(temp);
        }
}

int
PUMA::getNumberOfMeshChildren(ConnectivityListItem* connectivityList) {
        ConnectivityListItem* temp = connectivityList;
        int numberOfChildren = 0;
        while (temp != NULL) {
            if (
                (temp->meshMember()) &&

                ((temp->distanceToCore() - 1) > 
                (connectivityList->distanceToCore())) &&

                ((NOW - temp->lastMAReceived()) <= 
                (2 * MULTICAST_ANNOUNCEMENT)) &&

                ((ALL_SHORTEST_PATHS) || 
                 (temp->nextHopsNextHop() <= id_))
               )
                numberOfChildren++;
            temp = temp->nextItem_;
        }
        return numberOfChildren;
}

bool
PUMA::doIForwardThisPacket(MulticastGroup* group, Packet* p) {
        struct hdr_cmn   *ch = HDR_CMN(p);
        ConnectivityListItem* temp = group->connectivityList_->start_;
        while (temp != NULL) {
            if (ch->prev_hop_ == temp->nextHop())
            // has the forwarder used me as a next hop recently ? 
                if ( temp->lastMAReceived() != NEVER && 
                    ((NOW - temp->lastMAReceived()) <= 
                    (2 * MULTICAST_ANNOUNCEMENT)) && 
                    temp->nextHopsNextHop() == id_)
                    return true;
                else
                    return false;
            temp = temp->nextItem_;
	    }
        return false;
}

void
PUMA::handleDataFromTransport(Packet* p) {
        struct hdr_cmn   *ch = HDR_CMN(p);
        struct hdr_ip    *ih = HDR_IP(p);
        struct hdr_puma  *ph = HDR_PUMA(p);
        MulticastGroup* group = multicastGroups_.find(ih->daddr());
        if (group == NULL) {
            group = new MulticastGroup(ih->daddr(), INVALID_ADDRESS);
            multicastGroups_.append(group);
        }
        packetSequenceNumber_++;
        ch->ptype()          = PT_PUMA;