qospktio.c

Vista 核心Rally技术之-LLTD 实现源代码

                }

                bucket->evt_seqNum = g_sequencenum;
                bucket->evt_numEvts = 0;
            }

            evt = &bucket->evt_descs[bucket->evt_numEvts++];

            /* Copy timestamps, packet-ID, and set reserved-byte... */
            memcpy(&evt->ctrlr_txstamp, &g_qprb_hdr->probe_txstamp, sizeof(uint64_t));

            /* Pick up the rx-timestamp from the driver, put in global timestamp */
            cpy_hton64(&evt->sink_rxstamp, &g_pktio_timestamp);

            evt->evt_pktID = g_qprb_hdr->probe_pktID;
            evt->evt_reserved = 0;
        } while (FALSE);

        IF_TRACED(TRC_PACKET)
            dbgprintf("qos_timedprobe processed: seq=" FMT_UINT16 ", evtcount=" FMT_UINT32 "\n",
                      g_sequencenum, bucket->evt_numEvts);
        END_TRACE
    }
}


static
void
qos_query(void)
{
    qosEventBucket_t* bucket;
    qos_session_t*	pThisSsn = qos_find_session();
    size_t          nbytes;
    uint16_t        numEvts;

    /* Find associated session, and reject with silence if none found. */
    if (pThisSsn == NULL)   return;

    /* Valid session - mark it as "still active" */
    pThisSsn->qssn_ticks_til_discard = 4;

    /* Compose the response headers in the space left before the events descrs */
    /* Build the ethernet header and base header */
    fmt_base(g_txbuf, &g_hwaddr, &pThisSsn->qssn_ctrlr_real, ToS_QoSDiagnostics,
             Qopcode_QueryResp, g_sequencenum, FALSE /*no broadcast*/);
    nbytes = sizeof(topo_ether_header_t) + sizeof(topo_base_header_t);

    /* Locate the event bucket */
    bucket = qos_find_bucket(pThisSsn);

    /* Add the number of events */
    numEvts = bucket ? bucket->evt_numEvts : 0;
    g_short_reorder_buffer = htons(numEvts);
    memcpy(&g_txbuf[nbytes], &g_short_reorder_buffer, sizeof(uint16_t));
    nbytes += sizeof(uint16_t);

    /* Copy the events */
    if (numEvts)
    {
        memcpy(&g_txbuf[nbytes], bucket->evt_descs, sizeof(qosEventDescr_t) * numEvts);
        nbytes += sizeof(qosEventDescr_t) * numEvts;
    }

    /* And send it... */
    tx_write(g_txbuf, nbytes);

    IF_TRACED(TRC_PACKET)
        dbgprintf("qos_query_response: sending " FMT_UINT32 " events, seq=" FMT_UINT16 " -> " ETHERADDR_FMT "\n",
                  numEvts, g_sequencenum, ETHERADDR_PRINT(&pThisSsn->qssn_ctrlr_real));
    END_TRACE
}


static
void
qos_counterlease(void)
{
    struct timeval	now;

    IF_TRACED(TRC_QOS)
        dbgprintf("qos_counter_lease: timeout was: " FMT_UINT32 ", now: 300",g_samples_remaining);
    END_TRACE

    if (g_samples_remaining == 0)
    {
        /* start collection: repeats every second - until the lease runs out */
        gettimeofday(&now, NULL);
        now.tv_sec += 1;
        g_qos_CTA_timer = event_add(&now, interface_counter_recorder, /*state:*/NULL);
        IF_TRACED(TRC_QOS)
             dbgprintf("qos_counter_lease: 1-sec timer started\n");
        END_TRACE
    }
    g_samples_remaining = 300;
}


static
void
qos_snapshot(void)
{
    size_t		nbytes;
    uint8_t            *pMsg, sample_cnt;
    uint64_t            perf_timestamp;
    uint                count, next;
    qos_perf_sample    *pSample, *thisSample;

    /* Build the ethernet header and base header */
    pMsg = g_txbuf;
    fmt_base(pMsg, &g_hwaddr, &g_base_hdr->tbh_realsrc, ToS_QoSDiagnostics,
             Qopcode_CounterResult, g_sequencenum, FALSE /*no broadcast*/);
    nbytes = sizeof(topo_ether_header_t) + sizeof(topo_base_header_t);

    /* Calculate the sub-second interval and write the CounterResult header */
    stamp_time(&perf_timestamp);
    pMsg[nbytes++] = (uint8_t)(((perf_timestamp - g_perf_timestamp)*256)/1000000);      // sub-second span
    pMsg[nbytes++] = (uint8_t)BYTE_SCALE;
    pMsg[nbytes++] = (uint8_t)PKT_SCALE;
    sample_cnt = (g_snap_hdr->cnt_rqstd <= (uint8_t)g_sample_count) ? g_snap_hdr->cnt_rqstd : (uint8_t)g_sample_count;
    pMsg[nbytes++] = sample_cnt;

    IF_TRACED(TRC_QOS)
        dbgprintf("qos_snapshot: subsec: %d; sample-cnt: " FMT_UINT32 "\n",(int)pMsg[32], g_sample_count);
    END_TRACE

    /* Now copy the samples to the QosCounterResult msg */
    next = (g_next_sample + (60 - sample_cnt)) % 60;
    pSample = (qos_perf_sample*)&pMsg[nbytes];
    for (count=sample_cnt; count>0; count--)
    {
        thisSample = &g_perf_samples[next++];
        next %= 60;
        pSample->bytes_rcvd = htons(thisSample->bytes_rcvd);
        pSample->pkts_rcvd  = htons(thisSample->pkts_rcvd);
        pSample->bytes_sent = htons(thisSample->bytes_sent);
        pSample->pkts_sent  = htons(thisSample->pkts_sent);
        IF_TRACED(TRC_QOS)
            dbgprintf("  sample: rcvd: %d; r-pkts: %d;  sent: %d; s-pkts: %d\n",
                      thisSample->bytes_rcvd,thisSample->pkts_rcvd,thisSample->bytes_sent,thisSample->pkts_sent);
        END_TRACE
        pSample++;
        nbytes += sizeof(qos_perf_sample);
    }
    /* Now take a sub-second sample, and put it in the QosCounterResult msg */
    {
        uint32_t       rbytes, rpkts, tbytes, tpkts;
        uint32_t       delta1, delta2, delta3, delta4 ;

        /* save the counters, so normal sampling won't be disrupted */
        rbytes = g_rbytes; rpkts = g_rpkts; tbytes = g_tbytes; tpkts = g_tpkts;

        get_raw_samples();      // get current values for g_rbytes, g_rpkts, g_tbytes, g_tpkts

        delta1 = (g_rbytes-rbytes > 0)? (g_rbytes-rbytes) : 0;
        pSample->bytes_rcvd = htons(delta1 / BYTE_SCALE_FACTOR);

        delta2 = (g_rpkts-rpkts > 0)?   (g_rpkts-rpkts) : 0;
        pSample->pkts_rcvd  = htons(delta2 / PKT_SCALE_FACTOR);

        delta3 = (g_tbytes-tbytes > 0)? (g_tbytes-tbytes) : 0;
        pSample->bytes_sent = htons(delta3 / BYTE_SCALE_FACTOR);

        delta4 = (g_tpkts-tpkts > 0)?   (g_tpkts-tpkts) : 0;
        pSample->pkts_sent  = htons(delta4 / PKT_SCALE_FACTOR);

        IF_TRACED(TRC_QOS)
            dbgprintf("  sub-sec sample: rcvd: " FMT_UINT32 "; r-pkts: " FMT_UINT32 \
                      ";  sent: " FMT_UINT32 "; s-pkts: " FMT_UINT32 "\n",
                      delta1, delta2, delta3, delta4);
        END_TRACE

        /* restore the saved counters */
        g_rbytes = rbytes; g_rpkts = rpkts; g_tbytes = tbytes; g_tpkts = tpkts;

        /* count the subsecond sample in the packet length */
        nbytes += sizeof(qos_perf_sample);
    }

    /* And send it... */
    tx_write( g_txbuf, nbytes );

    IF_TRACED(TRC_PACKET)
        dbgprintf("qos_counter_result: sending " FMT_UINT32 " perf-samples + sub-sec-sample, seq=" FMT_UINT16 \
                  " -> " ETHERADDR_FMT "\n",
                  g_sample_count, g_sequencenum, ETHERADDR_PRINT(&g_base_hdr->tbh_realsrc));
    END_TRACE
}


/************************** Q O S   M E S S A G E   R E C E I V E R **************************/

/* Called by packetio_recv_handler() when msg ToS indicates QOS.
 * The ether and base headers are validated, and ether- and base-header ptrs are set up. */
extern void qosrcvpkt(void);

void
qosrcvpkt(void)
{
    uint16_t thisSeqnum;

    /* check global g_opcode for QOS case */
    if (g_opcode < 0 || g_opcode >= Qopcode_INVALID)
    {
    	warn("qospktrcv: g_opcode=%d is out of range for QoS msg; ignoring\n", g_opcode);
    	return;
    }
    IF_TRACED(TRC_PACKET)
        dbgprintf("QOS: g_opcode=%d\n",g_opcode);
    END_TRACE

    thisSeqnum = ntohs(g_base_hdr->tbh_seqnum);

    /* QosCounterLease frame must not be sequenced, everything else must be sequenced */
    /* QosCounterLease is the only one that's broadcasted */
    if (g_opcode != Qopcode_CounterLease)
    {
        if (thisSeqnum == 0)
        {
        	warn("qospktrcv: g_opcode=%d must be sequenced; ignoring\n", g_opcode);
            return;
        }
        else if (!ETHERADDR_EQUALS(&g_base_hdr->tbh_realdst, &g_hwaddr))
        {
        	warn("qospktrcv: g_opcode=%d must be directed; ignoring\n", g_opcode);
            return;
        }
    }
    else if (thisSeqnum != 0)
    {
    	warn("qospktrcv: QosCounterLease must not be sequenced; ignoring\n");
        return;
    }
    else if (!ETHERADDR_IS_BCAST(&g_base_hdr->tbh_realdst))
    {
    	warn("qospktrcv: QosCounterLease must be broadcasted; ignoring\n");
        return;
    }

    /* Validate source/dest real/ether addresses */
    if (!ETHERADDR_EQUALS(&g_ethernet_hdr->eh_src, &g_base_hdr->tbh_realsrc) ||
        !ETHERADDR_EQUALS(&g_ethernet_hdr->eh_dst, &g_base_hdr->tbh_realdst))
    {
        return;
    }

    /* print the frame */
    IF_TRACED(TRC_PACKET)
	dbgprintf(" [" ETHERADDR_FMT "] -> [" ETHERADDR_FMT "] %s (seq=%d)\n",
		ETHERADDR_PRINT(&g_ethernet_hdr->eh_src), ETHERADDR_PRINT(&g_ethernet_hdr->eh_dst),
		Qos_opcode_names[g_opcode], thisSeqnum);
    END_TRACE

    /* By this time, we are pretty sure the sequence number is valid, so save a global copy... */
    g_sequencenum = thisSeqnum;

    /* Set up global pointers to the 2 possible types of received headers that are bigger than the base hdr */
    g_qprb_hdr  = (qos_probe_header_t*)   (g_base_hdr + 1);
    g_qinit_hdr = (qos_initsink_header_t*)(g_base_hdr + 1);
    g_snap_hdr  = (qos_snapshot_header_t*)(g_base_hdr + 1);

    /* Finally, perform per-g_opcode validation & processing */
    switch (g_opcode)
    {
      case Qopcode_Probe:
        qos_probe();
        break;

      case Qopcode_Query:
        qos_query();
        break;

      case Qopcode_CounterSnapshot:
        qos_snapshot();
        break;

      case Qopcode_CounterLease:
        qos_counterlease();
        break;

      case Qopcode_InitializeSink:
        qos_initsink();
        break;

      case Qopcode_Reset:
        qos_reset();
        break;

      /* (invalid- or Sink-sent-packets are ignored completely) */
      case Qopcode_ACK:		 // Sent by Sink
      case Qopcode_QueryResp:	 // Sent by Sink
      case Qopcode_Ready:	 // Sent by Sink
      case Qopcode_Error:	 // Sent by Sink
      case Qopcode_CounterResult:// Sent by Sink
      case Opcode_INVALID:
      default:
        break;
    }
}