puma.cc

组播路由协议

        ch->direction()      = hdr_cmn::DOWN;
        ch->size()           += IP_HDR_LEN + PUMA_HDR_LEN;
        ch->error()          = 0;
        ch->prev_hop_        = id_;
        ch->next_hop()       = IP_BROADCAST;
        ch->addr_type()      = NS_AF_INET;
        ch->uid()            = packetSequenceNumber_;
        ih->saddr()          = id_;
        ih->sport()          = RT_PORT;
        ih->daddr()          = group->multicastAddress();
        ph->MACount()        = 0;
        
        if (amIMeshMember(group)) {
            MessageCacheItem* temp = new MessageCacheItem(ih->saddr(), 
                                                         packetSequenceNumber_);
            messageCache_.append(temp);
            Scheduler::instance().schedule(target_, p,
                                       BROADCAST_JITTER * random_.uniform(1.0));
        }
        else
            forwardFreshDataPacket(group, p);
}

void
PUMA::sendAPartitionConfirmationRequest(MulticastGroup* group) {
        nsaddr_t temp = group->coreId();
        group->coreId() = TENTATIVE_PARTITION;
        sendAGenericMA(group, PARTITION, temp);
        group->coreId() = temp;
}

void
PUMA::sendASingleMA(MulticastGroup* group) {
        sendAGenericMA(group, NORMAL, INVALID_ADDRESS);
}

void
PUMA::sendAGenericMA(MulticastGroup* group, int mode, nsaddr_t oldCore) {
        MulticastAnnouncement* MA = new MulticastAnnouncement;
        Packet* p = Packet::alloc(sizeof(struct MulticastAnnouncement));
        struct hdr_cmn   *ch = HDR_CMN(p);
        struct hdr_ip    *ih = HDR_IP(p);
        struct hdr_puma  *ph = HDR_PUMA(p);
        MA->multicastAddress_ = group->multicastAddress();
        MA->coreId_           = group->coreId();
        MA->distanceToCore_   = distanceToCore(group);
        MA->sequenceNumber_   = group->lastSequenceNumber();
        MA->meshMember_       = group->meshMember() = amIMeshMember(group);
        if (mode == NORMAL)
            MA->nextHop_      = getNextHop(group);
        else
            MA->nextHop_      = oldCore;
       
        ch->ptype()     = PT_PUMA;
        ch->direction() = hdr_cmn::DOWN;
        ch->size()      += IP_HDR_LEN + PUMA_HDR_LEN;
        ch->error()     = 0;
        ch->prev_hop_   = id_;
        ch->next_hop()  = IP_BROADCAST;
        ch->addr_type() = NS_AF_INET;
        ih->saddr()     = id_;
        ih->daddr()     = group->multicastAddress();
        ih->sport()     = RT_PORT;
        ih->dport()     = RT_PORT;
        ih->ttl()       = IP_DEF_TTL;
        ph->MACount()   = 1;

        memcpy((void*)(((PacketData*)p->userdata())->data()),(void*)MA,
                 sizeof(MulticastAnnouncement));
        Scheduler::instance().schedule(target_, p,
                                       BROADCAST_JITTER * random_.uniform(1.0));
}

void PUMA::handleMA(MulticastGroup* group, MulticastAnnouncement ma, 
                     nsaddr_t source) {
        ConnectivityListItem* temp;
        if (ma.coreId_ == INVALID_ADDRESS) {
            if (group != NULL && group->coreId() != INVALID_ADDRESS) {
                sendASingleMA(group);
            }
        }
        else if (ma.coreId_ == TENTATIVE_PARTITION) {
            if (distanceToCore(group) < ma.distanceToCore_)
                handlePartitionConfirmationRequest(group, ma.nextHop_, 
                                                       ma.sequenceNumber_);
        }
        else if (group->coreId() == INVALID_ADDRESS ||
                 ma.coreId_       > group->coreId()) {
                // Reconnection
            updateGroup(group, ma, source);
            sendASingleMA(group);
        }
        else if (ma.coreId_ != group->coreId() && 
            isPartitionFlagSet(group)) {
            // Partition
            updateGroup(group, ma, source);
            sendASingleMA(group);
        }
        else if (ma.coreId_ < group->coreId()) {
            // Useless MA.
        }
        else {
            bool meshMembershipAfter;
            if (ma.sequenceNumber_  > group->lastSequenceNumber())
                // Fresher MA.
                handleFresherMA(group, ma, source);
            else if (ma.sequenceNumber_ < group->lastSequenceNumber()) {
                temp = new ConnectivityListItem(source, 
                       ma.distanceToCore_ + 1, ma.sequenceNumber_, 
                       ma.nextHop_, ma.meshMember_);
                group->connectivityList_->addSorted(temp);
            }
            else if ((ma.distanceToCore_ + 1) < distanceToCore(group)) {
                temp = new ConnectivityListItem(source, 
                       ma.distanceToCore_ + 1, ma.sequenceNumber_, 
                       ma.nextHop_, ma.meshMember_);
                group->connectivityList_->addSorted(temp);
                if (isSingleGroupPresent())
                    sendADelayedMA(group);
            }
            else {
                temp = new ConnectivityListItem(source, 
                       ma.distanceToCore_ + 1, ma.sequenceNumber_, 
                       ma.nextHop_, ma.meshMember_);
                group->connectivityList_->addSorted(temp);
            }
            meshMembershipAfter = amIMeshMember(group);
            if (group->meshMember() != meshMembershipAfter &&
                isSingleGroupPresent())
                sendADelayedMA(group);
        }
}

void
PUMA::sendADelayedMA(MulticastGroup* group) {
        if (!(group->delayedMAPending())) {
            nsaddr_t* multicastAddress = new nsaddr_t();
            *multicastAddress = group->multicastAddress();
            group->delayedMAPending() = true;
            routingSetTimer(SEND_PENDING_MA, multicastAddress,
                                MULTICAST_ANNOUNCEMENT_DELAY);
        }
}

bool
PUMA::isSingleGroupPresent() {
        if (multicastGroups_.start_ == NULL)
            return false;
        else if (multicastGroups_.start_->nextItem_ == NULL)
            return true;
        else
            return false;
}

void
PUMA::handleFresherMA(MulticastGroup* group, MulticastAnnouncement ma, 
                       nsaddr_t source) {
        group->lastSequenceNumber() = ma.sequenceNumber_;
        group->lastMAReceived() = NOW;
        ConnectivityListItem* temp = new ConnectivityListItem(source, 
                                     ma.distanceToCore_ + 1, ma.sequenceNumber_,
                                     ma.nextHop_, ma.meshMember_);
        group->connectivityList_->addSorted(temp);
        if (isSingleGroupPresent() || group->partitionConfirmationWait())
            sendADelayedMA(group);
}

void
PUMA::handlePartitionConfirmationRequest(MulticastGroup* group, 
                                          nsaddr_t currentCore,
                                          unsigned int sequenceNumber) {
        if (group->coreId() == currentCore && 
            group->lastSequenceNumber() > sequenceNumber)
            sendASingleMA(group);
        else if (!group->partitionConfirmationWait())
            fightMultipleGroupElectionAfterPartition(group);
}

void
PUMA::sendNextBucket() {
        int x = multicastGroups_.getNumberOfGroups();
        routingSetTimer(SEND_NEXT_BUCKET, NULL, MULTICAST_ANNOUNCEMENT);
        MulticastGroup* group = multicastGroups_.start_;
        if (group != NULL && !isSingleGroupPresent() && x > 0) {
            Packet *p = Packet::alloc(sizeof(struct MulticastAnnouncement) * x);
            struct hdr_cmn   *ch = HDR_CMN(p);
            struct hdr_ip    *ih = HDR_IP(p);
            struct hdr_puma  *ph = HDR_PUMA(p);

            ch->ptype()     = PT_PUMA;
            ch->direction() = hdr_cmn::DOWN;
            ch->size()      += IP_HDR_LEN + PUMA_HDR_LEN;
            ch->error()     = 0;
            ch->prev_hop_   = id_;
            ch->addr_type() = NS_AF_INET;
            ch->next_hop()  = IP_BROADCAST;

            ih->saddr()     = id_;
            ih->daddr()     = group->multicastAddress();
            ih->sport()     = RT_PORT;
            ih->dport()     = RT_PORT;
            ih->ttl()       = IP_DEF_TTL;

            ph->MACount()        = x;

            struct MulticastAnnouncement bucket[x];
            int i = 0;
            MulticastGroup* temp = group;
            while (temp != NULL) {
                if (isPartitionFlagSet(temp) && temp->receiver())
	                if (!(temp->partitionConfirmationWait()))
		                fightMultipleGroupElectionAfterPartition(temp);
                bucket[i].multicastAddress_= temp->multicastAddress();
                bucket[i].coreId_          = temp->coreId();
                bucket[i].distanceToCore_  = distanceToCore(temp);
                bucket[i].sequenceNumber_  = temp->lastSequenceNumber();
                bucket[i].nextHop_         = getNextHop(temp);
                bucket[i].meshMember_      = temp->meshMember() = 
                                             amIMeshMember(temp);
                i++;
                temp = temp->nextItem_;
            }
            memcpy((void*)(((PacketData*)p->userdata())->data()),(void*)&bucket,
                 sizeof(MulticastAnnouncement) * x);
            Scheduler::instance().schedule(target_, p,
                                       BROADCAST_JITTER * random_.uniform(1.0));
        }
        else if (isSingleGroupPresent() && isPartitionFlagSet(group) && 
                group->receiver())
             becomeCoreAfterPartition(group);
}

void
PUMA::handleBucket(Packet* p) {
        struct hdr_ip    *ih = HDR_IP(p);
        struct hdr_puma *ph = HDR_PUMA(p);
        if (ih->saddr() == id_) {
            Packet::free(p);
            return;
        }
        recordACK(ih->saddr());
        struct MulticastAnnouncement bucket[ph->MACount()];
        memcpy((void*)&bucket,(void*)((PacketData*)p->userdata())->data(),
                sizeof(MulticastAnnouncement) * ph->MACount());
        MulticastGroup* group;
        nsaddr_t groupAddress;
        for (int i = 0; i < ph->MACount(); i++) {
            groupAddress = bucket[i].multicastAddress_;
            group = multicastGroups_.find(groupAddress);
            if (group == NULL) {
                group = new MulticastGroup(groupAddress, INVALID_ADDRESS);
                multicastGroups_.append(group);
            }
            handleMA(group, bucket[i], ih->saddr());
        }
}

void
PUMA::sendNextMA(MulticastGroup* group) {
        if (group != NULL && group->coreId() == id_) {
            bookKeepingBeforeMA(group);
            if (isSingleGroupPresent()) {
                sendASingleMA(group);
            }
        }
}

void
PUMA::bookKeepingBeforeMA(MulticastGroup* group) {
        nsaddr_t* multicastAddress =  new nsaddr_t();
        *multicastAddress = group->multicastAddress();
        routingSetTimer(SEND_NEXT_MA, (void*)multicastAddress,
                        MULTICAST_ANNOUNCEMENT);
        group->lastMAOriginated() = NOW;
        (group->lastSequenceNumber())++;
}

void
PUMA::recordACK(nsaddr_t source) {
        (ackTable_.numberOfACKTableEntries())++;
        if (ackTable_.numberOfACKTableEntries() > NUMBER_OF_ACK_TABLE_ENTRIES)
            ackTable_.numberOfACKTableEntries() = NUMBER_OF_ACK_TABLE_ENTRIES;
        ackTable_.ACKTableData_[ackTable_.
        nextEmptySlotInACKTable()].timeHeard() = NOW;
        ackTable_.ACKTableData_[ackTable_.
        nextEmptySlotInACKTable()].heardFrom() = source;
        (ackTable_.nextEmptySlotInACKTable())++;
        ackTable_.nextEmptySlotInACKTable() = ackTable_.nextEmptySlotInACKTable()
                                                  % NUMBER_OF_ACK_TABLE_ENTRIES;
}

int
PUMA::getNextHop(MulticastGroup* group) {
        if ((group == NULL) || (group->connectivityList_->start_ == NULL)) 
            return INVALID_ADDRESS;
        else if ((group->coreId() == id_))
            return INVALID_ADDRESS;
        else {
            ConnectivityListItem* temp = group->connectivityList_->start_;
            int i;
            for (i=1; i < NUMBER_OF_PARENTS; i++) {
                if (temp->nextItem_ == NULL)
                    break;
                if (temp->nextItem_->sequenceNumber() != 
                    group->connectivityList_->start_->sequenceNumber() ||
                    
                    temp->nextItem_->distanceToCore() !=
                    group->connectivityList_->start_->distanceToCore())
                    break;
                temp = temp->nextItem_;
            }
            return temp->nextHop();
        }
}

int
PUMA::distanceToCore(MulticastGroup* group) {
        if ((group != NULL) && (group->coreId() == id_))
            return 0;
        if ((group == NULL) || (group->connectivityList_->start_ == NULL))
            return INVALID_DISTANCE;
        return group->connectivityList_->start_->distanceToCore();
}