Add horizon attributes to fq

Add two attributes to fq: Horizon and HorizonDrop (which correspond to the
TCA_FQ_HORIZON and TCA_FQ_HORIZON_DROP attributes). The HorizonDrop attribute
specifies which policy to apply: drop (1, kernel default), cap delivery time to
horizon (0), and the Horizon attribute specifies the number of useconds before
applying the policy.

Add a new test TestFqHorizon to test the changes and as an example of usage.

Signed-off-by: Anton Protopopov <aspsk@isovalent.com>

nl/tc_linux.go

@@ -891,6 +891,10 @@ const (
 	TCA_FQ_FLOW_REFILL_DELAY  // flow credit refill delay in usec
 	TCA_FQ_ORPHAN_MASK        // mask applied to orphaned skb hashes
 	TCA_FQ_LOW_RATE_THRESHOLD // per packet delay under this rate
+	TCA_FQ_CE_THRESHOLD       // DCTCP-like CE-marking threshold
+	TCA_FQ_TIMER_SLACK        // timer slack
+	TCA_FQ_HORIZON            // time horizon in us
+	TCA_FQ_HORIZON_DROP       // drop packets beyond horizon, or cap their EDT
 )
 
 const (

qdisc.go

@@ -17,6 +17,12 @@ const (
 	HANDLE_MIN_EGRESS  = 0xFFFFFFF3
 )
 
+const (
+	HORIZON_DROP_POLICY_CAP     = 0
+	HORIZON_DROP_POLICY_DROP    = 1
+	HORIZON_DROP_POLICY_DEFAULT = 255
+)
+
 type Qdisc interface {
 	Attrs() *QdiscAttrs
 	Type() string
@@ -278,22 +284,25 @@ type Fq struct {
 	FlowDefaultRate uint32
 	FlowMaxRate     uint32
 	// called BucketsLog under the hood
-	Buckets          uint32
-	FlowRefillDelay  uint32
-	LowRateThreshold uint32
+	Buckets           uint32
+	FlowRefillDelay   uint32
+	LowRateThreshold  uint32
+	Horizon           uint32
+	HorizonDropPolicy uint8
 }
 
 func (fq *Fq) String() string {
 	return fmt.Sprintf(
-		"{PacketLimit: %v, FlowPacketLimit: %v, Quantum: %v, InitialQuantum: %v, Pacing: %v, FlowDefaultRate: %v, FlowMaxRate: %v, Buckets: %v, FlowRefillDelay: %v,  LowRateThreshold: %v}",
-		fq.PacketLimit, fq.FlowPacketLimit, fq.Quantum, fq.InitialQuantum, fq.Pacing, fq.FlowDefaultRate, fq.FlowMaxRate, fq.Buckets, fq.FlowRefillDelay, fq.LowRateThreshold,
+		"{PacketLimit: %v, FlowPacketLimit: %v, Quantum: %v, InitialQuantum: %v, Pacing: %v, FlowDefaultRate: %v, FlowMaxRate: %v, Buckets: %v, FlowRefillDelay: %v,  LowRateThreshold: %v, Horizon: %v, HorizonDropPolicy: %v}",
+		fq.PacketLimit, fq.FlowPacketLimit, fq.Quantum, fq.InitialQuantum, fq.Pacing, fq.FlowDefaultRate, fq.FlowMaxRate, fq.Buckets, fq.FlowRefillDelay, fq.LowRateThreshold, fq.Horizon, fq.HorizonDropPolicy,
 	)
 }
 
 func NewFq(attrs QdiscAttrs) *Fq {
 	return &Fq{
-		QdiscAttrs: attrs,
-		Pacing:     1,
+		QdiscAttrs:        attrs,
+		Pacing:            1,
+		HorizonDropPolicy: HORIZON_DROP_POLICY_DEFAULT,
 	}
 }
 

qdisc_linux.go

@@ -286,6 +286,12 @@ func qdiscPayload(req *nl.NetlinkRequest, qdisc Qdisc) error {
 		if qdisc.FlowDefaultRate > 0 {
 			options.AddRtAttr(nl.TCA_FQ_FLOW_DEFAULT_RATE, nl.Uint32Attr((uint32(qdisc.FlowDefaultRate))))
 		}
+		if qdisc.Horizon > 0 {
+			options.AddRtAttr(nl.TCA_FQ_HORIZON, nl.Uint32Attr(qdisc.Horizon))
+		}
+		if qdisc.HorizonDropPolicy != HORIZON_DROP_POLICY_DEFAULT {
+			options.AddRtAttr(nl.TCA_FQ_HORIZON_DROP, nl.Uint8Attr(qdisc.HorizonDropPolicy))
+		}
 	case *Sfq:
 		opt := nl.TcSfqQoptV1{}
 		opt.TcSfqQopt.Quantum = qdisc.Quantum
@@ -546,6 +552,11 @@ func parseFqData(qdisc Qdisc, data []syscall.NetlinkRouteAttr) error {
 			fq.FlowMaxRate = native.Uint32(datum.Value)
 		case nl.TCA_FQ_FLOW_DEFAULT_RATE:
 			fq.FlowDefaultRate = native.Uint32(datum.Value)
+		case nl.TCA_FQ_HORIZON:
+			fq.Horizon = native.Uint32(datum.Value)
+		case nl.TCA_FQ_HORIZON_DROP:
+			fq.HorizonDropPolicy = datum.Value[0]
+
 		}
 	}
 	return nil

qdisc_test.go

@@ -469,6 +469,63 @@ func TestFqAddChangeDel(t *testing.T) {
 	}
 }
 
+func TestFqHorizon(t *testing.T) {
+	minKernelRequired(t, 5, 7)
+
+	tearDown := setUpNetlinkTest(t)
+	defer tearDown()
+	if err := LinkAdd(&Ifb{LinkAttrs{Name: "foo"}}); err != nil {
+		t.Fatal(err)
+	}
+	link, err := LinkByName("foo")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err := LinkSetUp(link); err != nil {
+		t.Fatal(err)
+	}
+	qdisc := &Fq{
+		QdiscAttrs: QdiscAttrs{
+			LinkIndex: link.Attrs().Index,
+			Handle:    MakeHandle(1, 0),
+			Parent:    HANDLE_ROOT,
+		},
+		Horizon:           1000,
+		HorizonDropPolicy: HORIZON_DROP_POLICY_CAP,
+	}
+	if err := QdiscAdd(qdisc); err != nil {
+		t.Fatal(err)
+	}
+	qdiscs, err := SafeQdiscList(link)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(qdiscs) != 1 {
+		t.Fatal("Failed to add qdisc")
+	}
+	fq, ok := qdiscs[0].(*Fq)
+	if !ok {
+		t.Fatal("Qdisc is the wrong type")
+	}
+	if fq.Horizon != qdisc.Horizon {
+		t.Fatal("Horizon does not match")
+	}
+	if fq.HorizonDropPolicy != qdisc.HorizonDropPolicy {
+		t.Fatal("HorizonDropPolicy does not match")
+	}
+
+	if err := QdiscDel(qdisc); err != nil {
+		t.Fatal(err)
+	}
+	qdiscs, err = SafeQdiscList(link)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(qdiscs) != 0 {
+		t.Fatal("Failed to remove qdisc")
+	}
+}
+
 func TestFqCodelAddChangeDel(t *testing.T) {
 	minKernelRequired(t, 3, 4)