Merge pull request #55 from chantra/netem_qdisc

[netem] minimalist support for netem

class_test.go

@@ -79,6 +79,65 @@ func TestClassAddDel(t *testing.T) {
 	if htb.Cbuffer != class.Cbuffer {
 		t.Fatal("Cbuffer doesn't match")
 	}
+
+	qattrs := QdiscAttrs{
+		LinkIndex: link.Attrs().Index,
+		Handle:    MakeHandle(0x2, 0),
+		Parent:    MakeHandle(0xffff, 2),
+	}
+	nattrs := NetemQdiscAttrs{
+		Latency:     20000,
+		Loss:        23.4,
+		Duplicate:   14.3,
+		LossCorr:    8.34,
+		Jitter:      1000,
+		DelayCorr:   12.3,
+		ReorderProb: 23.4,
+		CorruptProb: 10.0,
+		CorruptCorr: 10,
+	}
+	qdiscnetem := NewNetem(qattrs, nattrs)
+	if err := QdiscAdd(qdiscnetem); err != nil {
+		t.Fatal(err)
+	}
+
+	qdiscs, err = QdiscList(link)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(qdiscs) != 2 {
+		t.Fatal("Failed to add qdisc")
+	}
+	_, ok = qdiscs[0].(*Htb)
+	if !ok {
+		t.Fatal("Qdisc is the wrong type")
+	}
+
+	netem, ok := qdiscs[1].(*Netem)
+	if !ok {
+		t.Fatal("Qdisc is the wrong type")
+	}
+	// Compare the record we got from the list with the one we created
+	if netem.Loss != qdiscnetem.Loss {
+		t.Fatal("Loss does not match")
+	}
+	if netem.Latency != qdiscnetem.Latency {
+		t.Fatal("Latency does not match")
+	}
+	if netem.CorruptProb != qdiscnetem.CorruptProb {
+		t.Fatal("CorruptProb does not match")
+	}
+	if netem.Jitter != qdiscnetem.Jitter {
+		t.Fatal("Jitter does not match")
+	}
+	if netem.LossCorr != qdiscnetem.LossCorr {
+		t.Fatal("Loss does not match")
+	}
+	if netem.DuplicateCorr != qdiscnetem.DuplicateCorr {
+		t.Fatal("DuplicateCorr does not match")
+	}
+
+	// Deletion
 	if err := ClassDel(class); err != nil {
 		t.Fatal(err)
 	}

nl/tc_linux.go

@@ -56,17 +56,21 @@ const (
 )
 
 const (
-	SizeofTcMsg       = 0x14
+	SizeofTcMsg          = 0x14
-	SizeofTcActionMsg = 0x04
+	SizeofTcActionMsg    = 0x04
-	SizeofTcPrioMap   = 0x14
+	SizeofTcPrioMap      = 0x14
-	SizeofTcRateSpec  = 0x0c
+	SizeofTcRateSpec     = 0x0c
-	SizeofTcTbfQopt   = 2*SizeofTcRateSpec + 0x0c
+	SizeofTcNetemQopt    = 0x18
-	SizeofTcHtbCopt   = 2*SizeofTcRateSpec + 0x14
+	SizeofTcNetemCorr    = 0x0c
-	SizeofTcHtbGlob   = 0x14
+	SizeofTcNetemReorder = 0x08
-	SizeofTcU32Key    = 0x10
+	SizeofTcNetemCorrupt = 0x08
-	SizeofTcU32Sel    = 0x10 // without keys
+	SizeofTcTbfQopt      = 2*SizeofTcRateSpec + 0x0c
-	SizeofTcMirred    = 0x1c
+	SizeofTcHtbCopt      = 2*SizeofTcRateSpec + 0x14
-	SizeofTcPolice    = 2*SizeofTcRateSpec + 0x20
+	SizeofTcHtbGlob      = 0x14
+	SizeofTcU32Key       = 0x10
+	SizeofTcU32Sel       = 0x10 // without keys
+	SizeofTcMirred       = 0x1c
+	SizeofTcPolice       = 2*SizeofTcRateSpec + 0x20
 )
 
 // struct tcmsg {
@@ -191,6 +195,121 @@ func (x *TcRateSpec) Serialize() []byte {
 	return (*(*[SizeofTcRateSpec]byte)(unsafe.Pointer(x)))[:]
 }
 
+/**
+* NETEM
+ */
+
+const (
+	TCA_NETEM_UNSPEC = iota
+	TCA_NETEM_CORR
+	TCA_NETEM_DELAY_DIST
+	TCA_NETEM_REORDER
+	TCA_NETEM_CORRUPT
+	TCA_NETEM_LOSS
+	TCA_NETEM_RATE
+	TCA_NETEM_ECN
+	TCA_NETEM_RATE64
+	TCA_NETEM_MAX = TCA_NETEM_RATE64
+)
+
+// struct tc_netem_qopt {
+//	__u32	latency;	/* added delay (us) */
+//	__u32   limit;		/* fifo limit (packets) */
+//	__u32	loss;		/* random packet loss (0=none ~0=100%) */
+//	__u32	gap;		/* re-ordering gap (0 for none) */
+//	__u32   duplicate;	/* random packet dup  (0=none ~0=100%) */
+// 	__u32	jitter;		/* random jitter in latency (us) */
+// };
+
+type TcNetemQopt struct {
+	Latency   uint32
+	Limit     uint32
+	Loss      uint32
+	Gap       uint32
+	Duplicate uint32
+	Jitter    uint32
+}
+
+func (msg *TcNetemQopt) Len() int {
+	return SizeofTcNetemQopt
+}
+
+func DeserializeTcNetemQopt(b []byte) *TcNetemQopt {
+	return (*TcNetemQopt)(unsafe.Pointer(&b[0:SizeofTcNetemQopt][0]))
+}
+
+func (x *TcNetemQopt) Serialize() []byte {
+	return (*(*[SizeofTcNetemQopt]byte)(unsafe.Pointer(x)))[:]
+}
+
+// struct tc_netem_corr {
+//  __u32   delay_corr; /* delay correlation */
+//  __u32   loss_corr;  /* packet loss correlation */
+//  __u32   dup_corr;   /* duplicate correlation  */
+// };
+
+type TcNetemCorr struct {
+	DelayCorr uint32
+	LossCorr  uint32
+	DupCorr   uint32
+}
+
+func (msg *TcNetemCorr) Len() int {
+	return SizeofTcNetemCorr
+}
+
+func DeserializeTcNetemCorr(b []byte) *TcNetemCorr {
+	return (*TcNetemCorr)(unsafe.Pointer(&b[0:SizeofTcNetemCorr][0]))
+}
+
+func (x *TcNetemCorr) Serialize() []byte {
+	return (*(*[SizeofTcNetemCorr]byte)(unsafe.Pointer(x)))[:]
+}
+
+// struct tc_netem_reorder {
+//  __u32   probability;
+//  __u32   correlation;
+// };
+
+type TcNetemReorder struct {
+	Probability uint32
+	Correlation uint32
+}
+
+func (msg *TcNetemReorder) Len() int {
+	return SizeofTcNetemReorder
+}
+
+func DeserializeTcNetemReorder(b []byte) *TcNetemReorder {
+	return (*TcNetemReorder)(unsafe.Pointer(&b[0:SizeofTcNetemReorder][0]))
+}
+
+func (x *TcNetemReorder) Serialize() []byte {
+	return (*(*[SizeofTcNetemReorder]byte)(unsafe.Pointer(x)))[:]
+}
+
+// struct tc_netem_corrupt {
+//  __u32   probability;
+//  __u32   correlation;
+// };
+
+type TcNetemCorrupt struct {
+	Probability uint32
+	Correlation uint32
+}
+
+func (msg *TcNetemCorrupt) Len() int {
+	return SizeofTcNetemCorrupt
+}
+
+func DeserializeTcNetemCorrupt(b []byte) *TcNetemCorrupt {
+	return (*TcNetemCorrupt)(unsafe.Pointer(&b[0:SizeofTcNetemCorrupt][0]))
+}
+
+func (x *TcNetemCorrupt) Serialize() []byte {
+	return (*(*[SizeofTcNetemCorrupt]byte)(unsafe.Pointer(x)))[:]
+}
+
 // struct tc_tbf_qopt {
 //   struct tc_ratespec rate;
 //   struct tc_ratespec peakrate;

qdisc.go

@@ -2,6 +2,7 @@ package netlink
 
 import (
 	"fmt"
+	"math"
 )
 
 const (
@@ -52,6 +53,14 @@ func HandleStr(handle uint32) string {
 	}
 }
 
+func Percentage2u32(percentage float32) uint32 {
+	// FIXME this is most likely not the best way to convert from % to uint32
+	if percentage == 100 {
+		return math.MaxUint32
+	}
+	return uint32(math.MaxUint32 * (percentage / 100))
+}
+
 // PfifoFast is the default qdisc created by the kernel if one has not
 // been defined for the interface
 type PfifoFast struct {
@@ -120,6 +129,120 @@ func (qdisc *Htb) Type() string {
 	return "htb"
 }
 
+// Netem is a classless qdisc that rate limits based on tokens
+
+type NetemQdiscAttrs struct {
+	Latency       uint32  // in us
+	DelayCorr     float32 // in %
+	Limit         uint32
+	Loss          float32 // in %
+	LossCorr      float32 // in %
+	Gap           uint32
+	Duplicate     float32 // in %
+	DuplicateCorr float32 // in %
+	Jitter        uint32  // in us
+	ReorderProb   float32 // in %
+	ReorderCorr   float32 // in %
+	CorruptProb   float32 // in %
+	CorruptCorr   float32 // in %
+}
+
+func (q NetemQdiscAttrs) String() string {
+	return fmt.Sprintf(
+		"{Latency: %d, Limit: %d, Loss: %d, Gap: %d, Duplicate: %d, Jitter: %d}",
+		q.Latency, q.Limit, q.Loss, q.Gap, q.Duplicate, q.Jitter,
+	)
+}
+
+type Netem struct {
+	QdiscAttrs
+	Latency       uint32
+	DelayCorr     uint32
+	Limit         uint32
+	Loss          uint32
+	LossCorr      uint32
+	Gap           uint32
+	Duplicate     uint32
+	DuplicateCorr uint32
+	Jitter        uint32
+	ReorderProb   uint32
+	ReorderCorr   uint32
+	CorruptProb   uint32
+	CorruptCorr   uint32
+}
+
+func NewNetem(attrs QdiscAttrs, nattrs NetemQdiscAttrs) *Netem {
+	var limit uint32 = 1000
+	var loss_corr, delay_corr, duplicate_corr uint32
+	var reorder_prob, reorder_corr uint32
+	var corrupt_prob, corrupt_corr uint32
+
+	latency := nattrs.Latency
+	loss := Percentage2u32(nattrs.Loss)
+	gap := nattrs.Gap
+	duplicate := Percentage2u32(nattrs.Duplicate)
+	jitter := nattrs.Jitter
+
+	// Correlation
+	if latency > 0 && jitter > 0 {
+		delay_corr = Percentage2u32(nattrs.DelayCorr)
+	}
+	if loss > 0 {
+		loss_corr = Percentage2u32(nattrs.LossCorr)
+	}
+	if duplicate > 0 {
+		duplicate_corr = Percentage2u32(nattrs.DuplicateCorr)
+	}
+	// FIXME should validate values(like loss/duplicate are percentages...)
+	latency = time2Tick(latency)
+
+	if nattrs.Limit != 0 {
+		limit = nattrs.Limit
+	}
+	// Jitter is only value if latency is > 0
+	if latency > 0 {
+		jitter = time2Tick(jitter)
+	}
+
+	reorder_prob = Percentage2u32(nattrs.ReorderProb)
+	reorder_corr = Percentage2u32(nattrs.ReorderCorr)
+
+	if reorder_prob > 0 {
+		// ERROR if lantency == 0
+		if gap == 0 {
+			gap = 1
+		}
+	}
+
+	corrupt_prob = Percentage2u32(nattrs.CorruptProb)
+	corrupt_corr = Percentage2u32(nattrs.CorruptCorr)
+
+	return &Netem{
+		QdiscAttrs:    attrs,
+		Latency:       latency,
+		DelayCorr:     delay_corr,
+		Limit:         limit,
+		Loss:          loss,
+		LossCorr:      loss_corr,
+		Gap:           gap,
+		Duplicate:     duplicate,
+		DuplicateCorr: duplicate_corr,
+		Jitter:        jitter,
+		ReorderProb:   reorder_prob,
+		ReorderCorr:   reorder_corr,
+		CorruptProb:   corrupt_prob,
+		CorruptCorr:   corrupt_corr,
+	}
+}
+
+func (qdisc *Netem) Attrs() *QdiscAttrs {
+	return &qdisc.QdiscAttrs
+}
+
+func (qdisc *Netem) Type() string {
+	return "netem"
+}
+
 // Tbf is a classless qdisc that rate limits based on tokens
 type Tbf struct {
 	QdiscAttrs

qdisc_linux.go

@@ -65,12 +65,45 @@ func QdiscAdd(qdisc Qdisc) error {
 		opt.DirectPkts = htb.DirectPkts
 		nl.NewRtAttrChild(options, nl.TCA_HTB_INIT, opt.Serialize())
 		// nl.NewRtAttrChild(options, nl.TCA_HTB_DIRECT_QLEN, opt.Serialize())
+	} else if netem, ok := qdisc.(*Netem); ok {
+		opt := nl.TcNetemQopt{}
+		opt.Latency = netem.Latency
+		opt.Limit = netem.Limit
+		opt.Loss = netem.Loss
+		opt.Gap = netem.Gap
+		opt.Duplicate = netem.Duplicate
+		opt.Jitter = netem.Jitter
+		options = nl.NewRtAttr(nl.TCA_OPTIONS, opt.Serialize())
+		// Correlation
+		corr := nl.TcNetemCorr{}
+		corr.DelayCorr = netem.DelayCorr
+		corr.LossCorr = netem.LossCorr
+		corr.DupCorr = netem.DuplicateCorr
+
+		if corr.DelayCorr > 0 || corr.LossCorr > 0 || corr.DupCorr > 0 {
+			nl.NewRtAttrChild(options, nl.TCA_NETEM_CORR, corr.Serialize())
+		}
+		// Corruption
+		corruption := nl.TcNetemCorrupt{}
+		corruption.Probability = netem.CorruptProb
+		corruption.Correlation = netem.CorruptCorr
+		if corruption.Probability > 0 {
+			nl.NewRtAttrChild(options, nl.TCA_NETEM_CORRUPT, corruption.Serialize())
+		}
+		// Reorder
+		reorder := nl.TcNetemReorder{}
+		reorder.Probability = netem.ReorderProb
+		reorder.Correlation = netem.ReorderCorr
+		if reorder.Probability > 0 {
+			nl.NewRtAttrChild(options, nl.TCA_NETEM_REORDER, reorder.Serialize())
+		}
 	} else if _, ok := qdisc.(*Ingress); ok {
 		// ingress filters must use the proper handle
 		if msg.Parent != HANDLE_INGRESS {
 			return fmt.Errorf("Ingress filters must set Parent to HANDLE_INGRESS")
 		}
 	}
+
 	req.AddData(options)
 	_, err := req.Execute(syscall.NETLINK_ROUTE, 0)
 	return err
@@ -135,6 +168,8 @@ func QdiscList(link Link) ([]Qdisc, error) {
 					qdisc = &Ingress{}
 				case "htb":
 					qdisc = &Htb{}
+				case "netem":
+					qdisc = &Netem{}
 				default:
 					qdisc = &GenericQdisc{QdiscType: qdiscType}
 				}
@@ -166,6 +201,10 @@ func QdiscList(link Link) ([]Qdisc, error) {
 					if err := parseHtbData(qdisc, data); err != nil {
 						return nil, err
 					}
+				case "netem":
+					if err := parseNetemData(qdisc, attr.Value); err != nil {
+						return nil, err
+					}
 
 					// no options for ingress
 				}
@@ -213,6 +252,40 @@ func parseHtbData(qdisc Qdisc, data []syscall.NetlinkRouteAttr) error {
 	}
 	return nil
 }
+
+func parseNetemData(qdisc Qdisc, value []byte) error {
+	netem := qdisc.(*Netem)
+	opt := nl.DeserializeTcNetemQopt(value)
+	netem.Latency = opt.Latency
+	netem.Limit = opt.Limit
+	netem.Loss = opt.Loss
+	netem.Gap = opt.Gap
+	netem.Duplicate = opt.Duplicate
+	netem.Jitter = opt.Jitter
+	data, err := nl.ParseRouteAttr(value[nl.SizeofTcNetemQopt:])
+	if err != nil {
+		return err
+	}
+	for _, datum := range data {
+		switch datum.Attr.Type {
+		case nl.TCA_NETEM_CORR:
+			opt := nl.DeserializeTcNetemCorr(datum.Value)
+			netem.DelayCorr = opt.DelayCorr
+			netem.LossCorr = opt.LossCorr
+			netem.DuplicateCorr = opt.DupCorr
+		case nl.TCA_NETEM_CORRUPT:
+			opt := nl.DeserializeTcNetemCorrupt(datum.Value)
+			netem.CorruptProb = opt.Probability
+			netem.CorruptCorr = opt.Correlation
+		case nl.TCA_NETEM_REORDER:
+			opt := nl.DeserializeTcNetemReorder(datum.Value)
+			netem.ReorderProb = opt.Probability
+			netem.ReorderCorr = opt.Correlation
+		}
+	}
+	return nil
+}
+
 func parseTbfData(qdisc Qdisc, data []syscall.NetlinkRouteAttr) error {
 	native = nl.NativeEndian()
 	tbf := qdisc.(*Tbf)

qdisc_test.go

@@ -88,13 +88,6 @@ func TestHtbAddDel(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	/*
-		cmd := exec.Command("tc", "qdisc")
-		out, err := cmd.CombinedOutput()
-		if err == nil {
-			fmt.Printf("%s\n", out)
-		}
-	*/
 	qdiscs, err := QdiscList(link)
 	if err != nil {
 		t.Fatal(err)
@@ -126,6 +119,7 @@ func TestHtbAddDel(t *testing.T) {
 		t.Fatal("Failed to remove qdisc")
 	}
 }
+
 func TestPrioAddDel(t *testing.T) {
 	tearDown := setUpNetlinkTest(t)
 	defer tearDown()