pback.c

Vitual Ring Routing 管你知不知道

            PO->Token.LocIF == SRP->Token.LocIF && 
            PO->Token.RemIF == SRP->Token.RemIF) {
        
            //
            // The opt can go in this SRP because they share the same NextHop.
            //
            uint OptionSize = sizeof(VRROption) + PO->Opt->Opt.optDataLen;

            //
            // Would it make the packet too big?
            // NB: Currently we check against PROTOCOL_MIN_FRAME_SIZE,
            // but sometimes we know which physical link will be used
            // and perhaps that physical link has a larger MTU.
            // Furthermore some options will not be forwarded.
            // So in some cases we could safely make the packet larger
            // than PROTOCOL_MIN_FRAME_SIZE.
            // gregos: allow space for MaintBufSendPacket() to insert an AR.
            // if (*Size + OptionSize > PROTOCOL_MIN_FRAME_SIZE) {
            if (*Size + OptionSize > PROTOCOL_MIN_FRAME_SIZE - sizeof(AcknowledgementRequest)) {
                goto KeepLooking;
            }

            //
            // Remove the option and add it to the packet.
            //
            *PrevPO = PO->Next;
            MsgMoveMessageToSRP(VA, PCache, SRP, PO->Opt);
            Count++;
            *Size += OptionSize;

            ExFreePool(PO);
        }
        else {
        KeepLooking:
            PrevPO = &PO->Next;
        }
    }

    return Count;
}

//* PbackPacketSize
//
//  Calculates the packet's current size, for the purposes of piggy-backing.
//  This makes worst-case assumptions about the possible growth in size
//  of some options.
//
uint
PbackPacketSize(SRPacket *SRP)
{
    uint Size;

    //
    // Start with the fixed-size VRR header.
    //
    Size = sizeof(VRRHeader);

    //
    // Add the size of the options.
    // This is exactly SROptionLength except
    // we assume worst-case size of Source Route
    // (because of Route Salvaging) and Route Request options.
    //
    Size += SROptionListLength((InternalOption *)SRP->ackreq);
    Size += SROptionListLength((InternalOption *)SRP->ack);
    Size += SROptionListLength((InternalOption *)SRP->Probe);
    Size += SROptionListLength((InternalOption *)SRP->ProbeReply);
    Size += SROptionListLength((InternalOption *)SRP->VrrHello);
    Size += SROptionListLength((InternalOption *)SRP->VrrSetup);
    Size += SROptionListLength((InternalOption *)SRP->VrrTearDown);
    Size += SROptionListLength((InternalOption *)SRP->VrrSetupReq);

    //
    // Add the size of the encrypted payload.
    //
    if (SRP->Packet != NULL) {
        uint Length;

        NdisQueryPacketLength(SRP->Packet, &Length);
        Size += Length;
    }

    return Size;
}

//* MsgLoadSRPfromList
//
//  Find messages from list than can be added to batch of
//  messages in a given SRP.
//
//  Should be called with the piggy-back cache locked,
//  if PrevPO is &PCache->List.
//
static uint
MsgLoadSRPfromList(
    PbackCache *PCache,
    PbackOption **PrevPO,       // Pointer to list of options.
    SRPacket *SRP)
{
    uint Size;
    uint Count;

    Size = PbackPacketSize(SRP);

    //
    // Add options destined for the packet's destination.
    //
    Count = MsgMoveMultiToSRP(PCache, PrevPO, SRP, &Size, SRP->Dest);

    return Count;
}

//* MsgLoadSRPfromPCache
//
//  Called when sending a packet. Checks if there are any
//  waiting options which should be added to the packet.
//
void
MsgLoadSRPfromPCache(
    MiniportAdapter *VA,
    SRPacket *SRP)
{
    PbackCache *PCache = &VA->PCache;
    uint Count;
    KIRQL OldIrql;

    KeAcquireSpinLock(&PCache->Lock, &OldIrql);
    Count = MsgLoadSRPfromList(PCache, &PCache->List, SRP);
    PCache->Number -= Count;
    KeReleaseSpinLock(&PCache->Lock, OldIrql);
}

//* PbackSendComplete
//
//  Completes the transmission of a control packet,
//  which is carrying options that could not be piggy-backed.
//
static void
PbackSendComplete(
    MiniportAdapter *VA,
    SRPacket *SRP,
    NDIS_STATUS Status)
{
    UNREFERENCED_PARAMETER(VA);
    UNREFERENCED_PARAMETER(Status);

    SRPacketFree(SRP);
}

//* MsgFailLink
//
//  Purge message queue of all messages for a given destination.
//
void
MsgFailLink(
    MiniportAdapter *VA,
    VirtualAddress Address,
    VRRIf LocIF,
    VRRIf RemIF)
{
    PbackCache *PCache = &VA->PCache;
    PbackOption *PO, **PrevPO = &PCache->List;
    KIRQL OldIrql;

    VrrTrace(VA,2,"MQ:**=FailLink", VA->Address,Address,Address,"LocIF",LocIF,"RemIF",RemIF);

    KeAcquireSpinLock(&PCache->Lock, &OldIrql);

    while ((PO = *PrevPO) != NULL) {
        if (VirtualAddressEqual(PO->Token.NextVAddress, Address) &&
            PO->Token.LocIF == LocIF && PO->Token.RemIF == RemIF) {

            if (PO->Opt->Opt.optionType == VRR_OPTION_TYPE_ACK) {
                Acknowledgement *Ack = (Acknowledgement *)&PO->Opt->Opt;
                VrrTrace(VA,3,"MQ:AK=Purj", Ack->from,Address,Ack->to,"id",Ack->identification, NULL, 0);
            }
            else if (PO->Opt->Opt.optionType == VRR_OPTION_TYPE_SETUP) {
                VrrSetup *Setup = (VrrSetup *)&PO->Opt->Opt;
                VrrTrace(VA,3,"MQ:SU=Purj" ,Setup->Source,Address,Setup->A,"Pid",RtlUlongByteSwap(Setup->PathId), NULL, 0);
            }
            else if (PO->Opt->Opt.optionType == VRR_OPTION_TYPE_TEARDOWN) {
                VrrTearDown *TD = (VrrTearDown *)&PO->Opt->Opt;
                int i, NumPId = RtlUlongByteSwap(TD->NumGlobalPathId);
                for (i = 0; i < NumPId; i++)
                    VrrTrace(VA,3,"MQ:TD=Purj", TD->Source,Address,TD->PId[i].Address,"Pid",RtlUlongByteSwap(TD->PId[i].PathId), NULL, 0);
            }
            else if (PO->Opt->Opt.optionType == VRR_OPTION_TYPE_SETUPREQ) {
                VrrSetupReq *SR = (VrrSetupReq *)&PO->Opt->Opt;
                VrrTrace(VA,3,"MQ:SR=Purj" ,SR->Source,Address,SR->Dest,"SeqNo",RtlUlongByteSwap(SR->FrameSeqNo), NULL, 0);
            }

            *PrevPO = PO->Next;
            ExFreePool(PO->Opt);
            ExFreePool(PO);
        }
        else {
            PrevPO = &PO->Next;
        }
    }

    KeReleaseSpinLock(&PCache->Lock, OldIrql);

}

//* MsgTimeoutPrint
//
//  Helper for MessageTimeout
//
void MsgTimeoutPrint(
    MiniportAdapter  *VA,
    InternalOption   *Opt,
    Time             Timeout,
    VirtualAddress   NextHop,
    uint             Wait)
{
    uint MicroSecs = TIME_TO_MS(Timeout) - TIMESTAMP;

    if (Opt->Opt.optionType == VRR_OPTION_TYPE_ACK) {
        Acknowledgement *Ack = (Acknowledgement *)&Opt->Opt;
        VrrTrace(VA, 3, Wait ? "MQ:AK=Wait" : "MQ:AK=Now_", Ack->from, NextHop, Ack->to,
               "SeqNo",Ack->identification, "Due", MicroSecs);
    }
    else if (Opt->Opt.optionType == VRR_OPTION_TYPE_SETUP) {
        VrrSetup *Setup = (VrrSetup *)&Opt->Opt;
        VrrTrace(VA, 3,Wait ? "MQ:SU=Wait" : "MQ:SU=Now_" ,Setup->Source, NextHop, Setup->A,
               "Pid",RtlUlongByteSwap(Setup->PathId), "Due", MicroSecs);
    }
    else if (Opt->Opt.optionType == VRR_OPTION_TYPE_TEARDOWN) {
        VrrTearDown *TD = (VrrTearDown *)&Opt->Opt;
        int i, NumPId = RtlUlongByteSwap(TD->NumGlobalPathId);
        for (i = 0; i < NumPId; i++)
            VrrTrace(VA, 3,Wait ? "MQ:TD=Wait" : "MQ:TD=Now_",TD->Source,NextHop,TD->PId[i].Address,
                   "Pid",RtlUlongByteSwap(TD->PId[i].PathId), "Due", MicroSecs);
    }
    else if (Opt->Opt.optionType == VRR_OPTION_TYPE_SETUPREQ) {
        VrrSetupReq *SR = (VrrSetupReq *)&Opt->Opt;
        VrrTrace(VA, 3,Wait ? "MQ:SR=Wait" : "MQ:SR=Now_",SR->Source,NextHop,SR->Dest,
               "FrameSeqNo",RtlUlongByteSwap(SR->FrameSeqNo), "Due", MicroSecs);
    }

}

//* MessageTimeout
//
//  Ensures that waiting options are sent by their timeout.
//
//  Called from MiniportTimeout at DPC level.
//
Time
MessageTimeout(
    MiniportAdapter *VA,
    Time Now)
{
    PbackCache *PCache = &VA->PCache;
    MaintBuf *MB = VA->MaintBuf;
    PbackOption *PO;
    PbackOption *PrevPO = NULL;
    Time Timeout = 0;
    uint NumMessages;
    Time MaxMsgDelay = 0;
    uint CountMsgDelayed = 0;

    //
    // Search the message queue for messages which are past their
    // transmission deadline and for which the maintenance buffer
    // is not blocked awaiting ack of an earlier packet.
    //
    // Note that the message queue is disturbed whenever a batch
    // of messages is taken from it.
    //
    KeAcquireSpinLockAtDpcLevel(&PCache->Lock);

    NumMessages = PCache->Number;
    
    for (PO = PCache->List; 
         PO != NULL; 
         PO = (PrevPO == NULL ? PCache->List : PrevPO->Next)) {

        MaintBufNode *MBN;
        SRPacket *SRP;
        uint BatchSize;
        Time Overdue;

        VRRASSERT(NumMessages > 0);
        if (NumMessages == 0)
            break;

        if (PO->Timeout > Now) {
            if (Timeout == 0)
                Timeout = PO->Timeout;
            MsgTimeoutPrint(VA, PO->Opt, PO->Timeout, PO->Token.NextVAddress,TRUE);
            PrevPO = PO;
            NumMessages--;
            continue;
        }

        KeAcquireSpinLockAtDpcLevel(&MB->Lock);
        MBN = MaintBufFindNode(MB, PO->Token.NextVAddress, 
                               PO->Token.LocIF, PO->Token.RemIF);
        KeReleaseSpinLockFromDpcLevel(&MB->Lock);
        
        if (MBN == NULL) {
            // Should never happen.
            MsgTimeoutPrint(VA, PO->Opt, PO->Timeout, PO->Token.NextVAddress, TRUE);
            PrevPO = PO;
            NumMessages--;
            continue;
        }

        //
        // If MBN is already has a (packet,ack) pair queued then wait
        // for it to complete; otherwise construct a batch of messages
        // in an SRP for the MBN to send.
        //
        if (MBN->MBP != NULL) {
            MsgTimeoutPrint(VA, PO->Opt, PO->Timeout, PO->Token.NextVAddress, TRUE);
            Timeout = Now + MAX_MSG_TIMOUT_INTERVAL;
            CountMsgDelayed++;
            if (MaxMsgDelay < Now - PO->Timeout)
                MaxMsgDelay = Now - PO->Timeout;
            PrevPO = PO;
            NumMessages--;
            continue;
        }

        //
        // Allocate an SRP for a batch of messages to this next hop.
        //
        SRP = ExAllocatePool(NonPagedPool, sizeof *SRP);
        if (SRP == NULL) {
            ExFreePool(SRP);
            VrrTrace(VA,2,"MQ:**=AllocErr",VA->Address,PO->Token.NextVAddress,PO->Dest,NULL,0,NULL,0);
            PrevPO = PO;
            NumMessages--;
            continue;
        }
        RtlZeroMemory(SRP, sizeof *SRP);

        //
        // Initialize SRP.
        //
        RtlCopyMemory(SRP->Dest, PO->Dest, SR_ADDR_LEN);
        RtlCopyMemory(SRP->Source, VA->Address, SR_ADDR_LEN);
        RtlCopyMemory(&SRP->Token, &PO->Token, sizeof(TxToken));
        RtlCopyMemory(SRP->Origin, VA->Address, SR_ADDR_LEN);
        SRP->FrameSeqNo = InterlockedIncrement((unsigned long *)&VrrGlobal.NextFrameSeqNo);
        SRP->HopCount = 0;

        //
        // Move a batch of messages into the SRP from the PCache.
        // Note: this call may invalidate PO->Next and PrevPO, so
        // start the next iteration at PCache->List which will
        // either be valid or NULL.
        //
        BatchSize = MsgLoadSRPfromList(PCache, &PCache->List, SRP);
        VRRASSERT(BatchSize != 0 && BatchSize <= PCache->Number);
        PCache->Number -= BatchSize;
        NumMessages = PCache->Number;
        PrevPO = NULL;

        //
        // Send the SRP using the maintenance buffer routines.
        //
        if (MaintBufSendPacket(VA, SRP, PbackSendComplete) == FALSE) {
            //
            // MaintBuf failed to send the packet, dropping the SRP and
            // the messages it contained. We must fail the link so that
            // the VRR protocol can recover to a consistent state.
            //
            // Note the handling of failing messages gets serialized on
            // the lock we are holding.
            //
            NeighborCacheEntry *NCE;

            VrrTrace(VA,2,"MQ:TO=FailLink",NULL,NULL,PO->Token.NextVAddress,"LocIF",PO->Token.LocIF,"RemIF",PO->Token.RemIF);
            KeAcquireSpinLockAtDpcLevel(&VA->NC.Lock);
            NCE=FindNCE(&VA->NC, PO->Token.NextVAddress, PO->Token.LocIF, PO->Token.RemIF, VRR_NCE_STATE_ANY);
            if (NCE != NULL)
                RouteFailLink(VA,NCE);
            KeReleaseSpinLockFromDpcLevel(&VA->NC.Lock);
        }
    }
    KeReleaseSpinLockFromDpcLevel(&PCache->Lock);

    //
    // Maintain counters on VA showing stalled message limits.
    //
    if ((uint)InterlockedCompareExchange(&VA->MaxMsgCountDelayed,0,0) < CountMsgDelayed)
        InterlockedExchange(&VA->MaxMsgCountDelayed, CountMsgDelayed );
    if ((uint)InterlockedCompareExchange(&VA->MaxMsgTimeDelay,0,0) < TIME_TO_MS(MaxMsgDelay))
        InterlockedExchange(&VA->MaxMsgTimeDelay, TIME_TO_MS(MaxMsgDelay));

    if (Timeout == 0)
        Timeout = MAXTIME;

    return Timeout;
}