FreeBSD/sys/netgraph/ng_pipe.c
/*
 * Dequeueing sequence - we basically do the following:
 *  1) Try to extract the frame from the inbound (bandwidth) queue;
 *  2) In accordance to BER specified, discard the frame randomly;
 *  3) If the frame survives BER, prepend it with delay info and move it
 *     to outbound (delay) queue;
 *  4) Loop to 2) until bandwidth quota for this timeslice is reached, or
 *     inbound queue is flushed completely;
 *  5) Dequeue frames from the outbound queue and send them downstream until
 *     outbound queue is flushed completely, or the next frame in the queue
 *     is not due to be dequeued yet
 */
Dequeueing sequence - we basically do the following:
1) Try to extract the frame from the inbound (bandwidth) queue;
2) In accordance to BER specified, discard the frame randomly;
3) If the frame survives BER, prepend it with delay info and move it
to outbound (delay) queue;
4) Loop to 2) until bandwidth quota for this timeslice is reached, or
inbound queue is flushed completely;
5) Dequeue frames from the outbound queue and send them downstream until
outbound queue is flushed completely, or the next frame in the queue
is not due to be dequeued yet
Dequeueing sequence - we basically do the following : 1 ) Try to extract the frame from the inbound ( bandwidth ) queue ; 2 ) In accordance to BER specified discard the frame randomly ; 3 ) If the frame survives BER prepend it with delay info and move it to outbound ( delay ) queue ; 4 ) Loop to 2 ) until bandwidth quota for this timeslice is reached or inbound queue is flushed completely ; 5 ) Dequeue frames from the outbound queue and send them downstream until outbound queue is flushed completely or the next frame in the queue is not due to be dequeued yet
static void
pipe_dequeue(struct hookinfo *hinfo, struct timeval *now) {
	static uint64_t rand, oldrand;
	const node_p node = NG_HOOK_NODE(hinfo->hook);
	const priv_p priv = NG_NODE_PRIVATE(node);
	struct hookinfo *dest;
	struct ngp_fifo *ngp_f, *ngp_f1;
	struct ngp_hdr *ngp_h;
	struct timeval *when;
	struct mbuf *m;
	int plen, error = 0;
	if (hinfo == &priv->lower)
		dest = &priv->upper;
	else
		dest = &priv->lower;
	while ((ngp_f = TAILQ_FIRST(&hinfo->fifo_head))) {
		when = &hinfo->qin_utime;
		if (when->tv_sec > now->tv_sec || (when->tv_sec == now->tv_sec
		    && when->tv_usec > now->tv_usec))
			break;
		ngp_h = TAILQ_FIRST(&ngp_f->packet_head);
		m = ngp_h->m;
		if (hinfo->cfg.drr) {
			if (ngp_f->rr_deficit >= m->m_pkthdr.len) {
				ngp_f->rr_deficit -= m->m_pkthdr.len;
			} else {
				ngp_f->rr_deficit += hinfo->cfg.drr;
				TAILQ_REMOVE(&hinfo->fifo_head, ngp_f, fifo_le);
				TAILQ_INSERT_TAIL(&hinfo->fifo_head,
				    ngp_f, fifo_le);
				continue;
			}
		}
		if (hinfo->cfg.duplicate &&
		    random() % 100 <= hinfo->cfg.duplicate) {
			ngp_h = uma_zalloc(ngp_zone, M_NOWAIT);
			KASSERT(ngp_h != NULL, ("ngp_h zalloc failed (3)"));
			m = m_dup(m, M_NOWAIT);
			KASSERT(m != NULL, ("m_dup failed"));
			ngp_h->m = m;
		} else {
			TAILQ_REMOVE(&ngp_f->packet_head, ngp_h, ngp_link);
			hinfo->run.qin_frames--;
			hinfo->run.qin_octets -= m->m_pkthdr.len;
			ngp_f->packets--;
		}
		if (hinfo->cfg.bandwidth) {
			hinfo->qin_utime.tv_usec +=
			    ((uint64_t) m->m_pkthdr.len + priv->overhead ) *
			    8000000 / hinfo->cfg.bandwidth;
			hinfo->qin_utime.tv_sec +=
			    hinfo->qin_utime.tv_usec / 1000000;
			hinfo->qin_utime.tv_usec =
			    hinfo->qin_utime.tv_usec % 1000000;
		}
		when = &ngp_h->when;
		when->tv_sec = hinfo->qin_utime.tv_sec;
		when->tv_usec = hinfo->qin_utime.tv_usec;
		if (ngp_f->packets) {
			if (hinfo->cfg.wfq) {
				TAILQ_REMOVE(&hinfo->fifo_head, ngp_f, fifo_le);
				FIFO_VTIME_SORT(TAILQ_FIRST(
				    &ngp_f->packet_head)->m->m_pkthdr.len)
			}
		} else {
			TAILQ_REMOVE(&hinfo->fifo_head, ngp_f, fifo_le);
			uma_zfree(ngp_zone, ngp_f);
			hinfo->run.fifo_queues--;
		}
		if (hinfo->cfg.ber) {
			oldrand = rand;
			rand = random();
			if (((oldrand ^ rand) << 17) >=
			    hinfo->ber_p[priv->overhead + m->m_pkthdr.len]) {
				hinfo->stats.out_disc_frames++;
				hinfo->stats.out_disc_octets += m->m_pkthdr.len;
				uma_zfree(ngp_zone, ngp_h);
				m_freem(m);
				continue;
			}
		}
		if (hinfo->cfg.qout_size_limit &&
		    hinfo->run.qout_frames>=hinfo->cfg.qout_size_limit) {
			hinfo->stats.out_disc_frames++;
			hinfo->stats.out_disc_octets += m->m_pkthdr.len;
			uma_zfree(ngp_zone, ngp_h);
			m_freem(m);
			continue;
		}
		when->tv_usec += priv->delay;
		when->tv_sec += when->tv_usec / 1000000;
		when->tv_usec = when->tv_usec % 1000000;
		TAILQ_INSERT_TAIL(&hinfo->qout_head, ngp_h, ngp_link);
		hinfo->run.qout_frames++;
		hinfo->run.qout_octets += m->m_pkthdr.len;
	}
	while ((ngp_h = TAILQ_FIRST(&hinfo->qout_head))) {
		when = &ngp_h->when;
		m = ngp_h->m;
		if (when->tv_sec > now->tv_sec ||
		    (when->tv_sec == now->tv_sec &&
		    when->tv_usec > now->tv_usec))
			break;
		plen = m->m_pkthdr.len;
		hinfo->run.qout_frames--;
		hinfo->run.qout_octets -= plen;
		TAILQ_REMOVE(&hinfo->qout_head, ngp_h, ngp_link);
		uma_zfree(ngp_zone, ngp_h);
		NG_SEND_DATA(error, dest->hook, m, meta);
		if (error) {
			hinfo->stats.out_disc_frames++;
			hinfo->stats.out_disc_octets += plen;
		} else {
			hinfo->stats.fwd_frames++;
			hinfo->stats.fwd_octets += plen;
		}
	}
	if ((hinfo->run.qin_frames != 0 || hinfo->run.qout_frames != 0) &&
	    !priv->timer_scheduled) {
		ng_callout(&priv->timer, node, NULL, 1, ngp_callout, NULL, 0);
		priv->timer_scheduled = 1;
	}
}
static void pipe_dequeue ( struct hookinfo * hinfo , struct timeval * now ) { static uint64_t rand , oldrand ; const node_p node = NG_HOOK_NODE ( hinfo -> hook ) ; const priv_p priv = NG_NODE_PRIVATE ( node ) ; struct hookinfo * dest ; struct ngp_fifo * ngp_f , * ngp_f1 ; struct ngp_hdr * ngp_h ; struct timeval * when ; struct mbuf * m ; int plen , error = 0 ; if ( hinfo == & priv -> lower ) dest = & priv -> upper ; else dest = & priv -> lower ; while ( ( ngp_f = TAILQ_FIRST ( & hinfo -> fifo_head ) ) ) { when = & hinfo -> qin_utime ; if ( when -> tv_sec > now -> tv_sec || ( when -> tv_sec == now -> tv_sec && when -> tv_usec > now -> tv_usec ) ) break ; ngp_h = TAILQ_FIRST ( & ngp_f -> packet_head ) ; m = ngp_h -> m ; if ( hinfo -> cfg . drr ) { if ( ngp_f -> rr_deficit >= m -> m_pkthdr . len ) { ngp_f -> rr_deficit -= m -> m_pkthdr . len ; } else { ngp_f -> rr_deficit += hinfo -> cfg . drr ; TAILQ_REMOVE ( & hinfo -> fifo_head , ngp_f , fifo_le ) ; TAILQ_INSERT_TAIL ( & hinfo -> fifo_head , ngp_f , fifo_le ) ; continue ; } } if ( hinfo -> cfg . duplicate && random ( ) % 100 <= hinfo -> cfg . duplicate ) { ngp_h = uma_zalloc ( ngp_zone , M_NOWAIT ) ; KASSERT ( ngp_h != NULL , ( " " ) ) ; m = m_dup ( m , M_NOWAIT ) ; KASSERT ( m != NULL , ( " " ) ) ; ngp_h -> m = m ; } else { TAILQ_REMOVE ( & ngp_f -> packet_head , ngp_h , ngp_link ) ; hinfo -> run . qin_frames -- ; hinfo -> run . qin_octets -= m -> m_pkthdr . len ; ngp_f -> packets -- ; } if ( hinfo -> cfg . bandwidth ) { hinfo -> qin_utime . tv_usec += ( ( uint64_t ) m -> m_pkthdr . len + priv -> overhead ) * 8000000 / hinfo -> cfg . bandwidth ; hinfo -> qin_utime . tv_sec += hinfo -> qin_utime . tv_usec / 1000000 ; hinfo -> qin_utime . tv_usec = hinfo -> qin_utime . tv_usec % 1000000 ; } when = & ngp_h -> when ; when -> tv_sec = hinfo -> qin_utime . tv_sec ; when -> tv_usec = hinfo -> qin_utime . tv_usec ; if ( ngp_f -> packets ) { if ( hinfo -> cfg . wfq ) { TAILQ_REMOVE ( & hinfo -> fifo_head , ngp_f , fifo_le ) ; FIFO_VTIME_SORT ( TAILQ_FIRST ( & ngp_f -> packet_head ) -> m -> m_pkthdr . len ) } } else { TAILQ_REMOVE ( & hinfo -> fifo_head , ngp_f , fifo_le ) ; uma_zfree ( ngp_zone , ngp_f ) ; hinfo -> run . fifo_queues -- ; } if ( hinfo -> cfg . ber ) { oldrand = rand ; rand = random ( ) ; if ( ( ( oldrand ^ rand ) << 17 ) >= hinfo -> ber_p [ priv -> overhead + m -> m_pkthdr . len ] ) { hinfo -> stats . out_disc_frames ++ ; hinfo -> stats . out_disc_octets += m -> m_pkthdr . len ; uma_zfree ( ngp_zone , ngp_h ) ; m_freem ( m ) ; continue ; } } if ( hinfo -> cfg . qout_size_limit && hinfo -> run . qout_frames >= hinfo -> cfg . qout_size_limit ) { hinfo -> stats . out_disc_frames ++ ; hinfo -> stats . out_disc_octets += m -> m_pkthdr . len ; uma_zfree ( ngp_zone , ngp_h ) ; m_freem ( m ) ; continue ; } when -> tv_usec += priv -> delay ; when -> tv_sec += when -> tv_usec / 1000000 ; when -> tv_usec = when -> tv_usec % 1000000 ; TAILQ_INSERT_TAIL ( & hinfo -> qout_head , ngp_h , ngp_link ) ; hinfo -> run . qout_frames ++ ; hinfo -> run . qout_octets += m -> m_pkthdr . len ; } while ( ( ngp_h = TAILQ_FIRST ( & hinfo -> qout_head ) ) ) { when = & ngp_h -> when ; m = ngp_h -> m ; if ( when -> tv_sec > now -> tv_sec || ( when -> tv_sec == now -> tv_sec && when -> tv_usec > now -> tv_usec ) ) break ; plen = m -> m_pkthdr . len ; hinfo -> run . qout_frames -- ; hinfo -> run . qout_octets -= plen ; TAILQ_REMOVE ( & hinfo -> qout_head , ngp_h , ngp_link ) ; uma_zfree ( ngp_zone , ngp_h ) ; NG_SEND_DATA ( error , dest -> hook , m , meta ) ; if ( error ) { hinfo -> stats . out_disc_frames ++ ; hinfo -> stats . out_disc_octets += plen ; } else { hinfo -> stats . fwd_frames ++ ; hinfo -> stats . fwd_octets += plen ; } } if ( ( hinfo -> run . qin_frames != 0 || hinfo -> run . qout_frames != 0 ) && ! priv -> timer_scheduled ) { ng_callout ( & priv -> timer , node , NULL , 1 , ngp_callout , NULL , 0 ) ; priv -> timer_scheduled = 1 ; } }
Dequeueing sequence - we basically do the following:
  1) Try to extract the frame from the inbound (bandwidth) queue;
  2) In accordance to BER specified, discard the frame randomly;
  3) If the frame survives BER, prepend it with delay info and move it
     to outbound (delay) queue;
  4) Loop to 2) until bandwidth quota for this timeslice is reached, or
     inbound queue is flushed completely;
  5) Dequeue frames from the outbound queue and send them downstream until
     outbound queue is flushed completely, or the next frame in the queue
     is not due to be dequeued yet
Dequeueing sequence - we basically do the following : 1 ) Try to extract the frame from the inbound ( bandwidth ) queue ; 2 ) In accordance to BER specified discard the frame randomly ; 3 ) If the frame survives BER prepend it with delay info and move it to outbound ( delay ) queue ; 4 ) Loop to 2 ) until bandwidth quota for this timeslice is reached or inbound queue is flushed completely ; 5 ) Dequeue frames from the outbound queue and send them downstream until outbound queue is flushed completely or the next frame in the queue is not due to be dequeued yet
/* Which one is the destination hook? */ /* Bandwidth queue processing */ /* Deficit Round Robin (DRR) processing */ /*
		 * Either create a duplicate and pass it on, or dequeue
		 * the original packet...
		 */ /* Calculate the serialization delay */ /* Sort / rearrange inbound queues */ /* Randomly discard the frame, according to BER setting */ /* Discard frame if outbound queue size limit exceeded */ /* Calculate the propagation delay */ /* Put the frame into the delay queue */ /* Delay queue processing */ /* Update outbound queue stats */ /* Dequeue the packet from qout */