From 02e1f2bb4558f7e162e23606ade4d2defa0868e1 Mon Sep 17 00:00:00 2001 From: Nate Hurley Date: Tue, 19 Apr 2022 09:15:18 -0700 Subject: [PATCH] Implement the initial multicast routing table. MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The purpose of this change is to: - Introduce the needed abstractions for the table. - Implement support for queuing pending packets. Subsequent changes will: - Implement the other needed table operations (i.e. AddInstalledRoute, RemoveInstalledRoute, and GetLastUsedTimestamp). - Implement the logic for expiring pending packets. Updates #7338. PiperOrigin-RevId: 442828835 --- pkg/tcpip/network/internal/multicast/BUILD | 50 +++ .../internal/multicast/example_test.go | 101 ++++++ .../network/internal/multicast/route_table.go | 307 ++++++++++++++++++ .../internal/multicast/route_table_test.go | 186 +++++++++++ 4 files changed, 644 insertions(+) create mode 100644 pkg/tcpip/network/internal/multicast/BUILD create mode 100644 pkg/tcpip/network/internal/multicast/example_test.go create mode 100644 pkg/tcpip/network/internal/multicast/route_table.go create mode 100644 pkg/tcpip/network/internal/multicast/route_table_test.go diff --git a/pkg/tcpip/network/internal/multicast/BUILD b/pkg/tcpip/network/internal/multicast/BUILD new file mode 100644 index 000000000..e274d4f37 --- /dev/null +++ b/pkg/tcpip/network/internal/multicast/BUILD @@ -0,0 +1,50 @@ +load("//tools:defs.bzl", "go_library", "go_test") + +package(licenses = ["notice"]) + +go_library( + name = "multicast", + srcs = [ + "route_table.go", + ], + visibility = ["//visibility:public"], + deps = [ + "//pkg/atomicbitops", + "//pkg/tcpip", + "//pkg/tcpip/stack", + ], +) + +go_test( + name = "multicast_test", + size = "small", + srcs = ["route_table_test.go"], + library = ":multicast", + deps = [ + "//pkg/atomicbitops", + "//pkg/refs", + "//pkg/refsvfs2", + "//pkg/tcpip", + "//pkg/tcpip/buffer", + "//pkg/tcpip/faketime", + "//pkg/tcpip/stack", + "//pkg/tcpip/testutil", + "@com_github_google_go_cmp//cmp:go_default_library", + "@com_github_google_go_cmp//cmp/cmpopts:go_default_library", + ], +) + +go_test( + name = "multicast_x_test", + size = "small", + srcs = ["example_test.go"], + deps = [ + ":multicast", + "//pkg/refs", + "//pkg/refsvfs2", + "//pkg/tcpip/buffer", + "//pkg/tcpip/faketime", + "//pkg/tcpip/stack", + "//pkg/tcpip/testutil", + ], +) diff --git a/pkg/tcpip/network/internal/multicast/example_test.go b/pkg/tcpip/network/internal/multicast/example_test.go new file mode 100644 index 000000000..6b05b41ce --- /dev/null +++ b/pkg/tcpip/network/internal/multicast/example_test.go @@ -0,0 +1,101 @@ +// Copyright 2022 The gVisor Authors. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package multicast_test + +import ( + "fmt" + "os" + "testing" + + "gvisor.dev/gvisor/pkg/refs" + "gvisor.dev/gvisor/pkg/refsvfs2" + "gvisor.dev/gvisor/pkg/tcpip/buffer" + "gvisor.dev/gvisor/pkg/tcpip/faketime" + "gvisor.dev/gvisor/pkg/tcpip/network/internal/multicast" + "gvisor.dev/gvisor/pkg/tcpip/stack" + "gvisor.dev/gvisor/pkg/tcpip/testutil" +) + +// Example shows how to interact with a multicast RouteTable. +func Example() { + address := testutil.MustParse4("192.168.1.1") + routeKey := multicast.RouteKey{UnicastSource: address, MulticastDestination: address} + + pkt := newPacketBuffer("hello") + defer pkt.DecRef() + + // Create a route table from a specified config. + table := multicast.RouteTable{} + config := multicast.DefaultConfig(faketime.NewManualClock()) + + if err := table.Init(config); err != nil { + panic(err) + } + + // Each entry in the table represents either an installed route or a pending + // route. To insert a pending route, call: + result, err := table.GetRouteOrInsertPending(routeKey, pkt) + + // Callers should handle a no buffer space error (e.g. only deliver the + // packet locally). + if err == multicast.ErrNoBufferSpace { + deliverPktLocally(pkt) + } + + if err != nil { + panic(err) + } + + // Callers should handle the various pending route states. + switch result.PendingRouteState { + case multicast.PendingRouteStateNone: + // The packet can be forwarded using the installed route. + forwardPkt(pkt, result.InstalledRoute) + case multicast.PendingRouteStateInstalled: + // The route has just entered the pending state. + emitMissingRouteEvent(routeKey) + deliverPktLocally(pkt) + case multicast.PendingRouteStateAppended: + // The route was already in the pending state. + deliverPktLocally(pkt) + } + + // Output: + // emitMissingRouteEvent + // deliverPktLocally +} + +func forwardPkt(*stack.PacketBuffer, *multicast.InstalledRoute) {} + +func emitMissingRouteEvent(multicast.RouteKey) { + fmt.Println("emitMissingRouteEvent") +} + +func deliverPktLocally(*stack.PacketBuffer) { + fmt.Println("deliverPktLocally") +} + +func newPacketBuffer(body string) *stack.PacketBuffer { + return stack.NewPacketBuffer(stack.PacketBufferOptions{ + Data: buffer.View(body).ToVectorisedView(), + }) +} + +func TestMain(m *testing.M) { + refs.SetLeakMode(refs.LeaksPanic) + code := m.Run() + refsvfs2.DoLeakCheck() + os.Exit(code) +} diff --git a/pkg/tcpip/network/internal/multicast/route_table.go b/pkg/tcpip/network/internal/multicast/route_table.go new file mode 100644 index 000000000..96e104c41 --- /dev/null +++ b/pkg/tcpip/network/internal/multicast/route_table.go @@ -0,0 +1,307 @@ +// Copyright 2022 The gVisor Authors. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +// Package multicast contains utilities for supporting multicast routing. +package multicast + +import ( + "errors" + "fmt" + "sync" + "time" + + "gvisor.dev/gvisor/pkg/atomicbitops" + "gvisor.dev/gvisor/pkg/tcpip" + "gvisor.dev/gvisor/pkg/tcpip/stack" +) + +// RouteTable represents a multicast routing table. +type RouteTable struct { + // Internally, installed and pending routes are stored and locked separately + // A couple of reasons for structuring the table this way: + // + // 1. We can avoid write locking installed routes when pending packets are + // being queued. In other words, the happy path of reading installed + // routes doesn't require an exclusive lock. + // 2. The cleanup process for expired routes only needs to operate on pending + // routes. Like above, a write lock on the installed routes can be + // avoided. + // 3. This structure is similar to the Linux implementation: + // https://github.com/torvalds/linux/blob/cffb2b72d3e/include/linux/mroute_base.h#L250 + + // TODO(https://gvisor.dev/issue/7338): Implement time based expiration of + // pending packets. + + // The installedMu lock should typically be acquired before the pendingMu + // lock. This ensures that installed routes can continue to be read even when + // the pending routes are write locked. + + installedMu sync.RWMutex + // Maintaining pointers ensures that the installed routes are exclusively + // locked only when a route is being installed. + // +checklocks:installedMu + installedRoutes map[RouteKey]*InstalledRoute + + pendingMu sync.RWMutex + // +checklocks:pendingMu + pendingRoutes map[RouteKey]pendingRoute + + config Config +} + +var ( + // ErrNoBufferSpace indicates that no buffer space is available in the + // pending route packet queue. + ErrNoBufferSpace = errors.New("unable to queue packet, no buffer space available") + + // ErrMissingClock indicates that a clock was not provided as part of the + // Config, but is required. + ErrMissingClock = errors.New("clock must not be nil") + + // ErrAlreadyInitialized indicate that RouteTable.Init was already invoked. + ErrAlreadyInitialized = errors.New("table is already initialized") +) + +// RouteKey represents an entry key in the RouteTable. +type RouteKey struct { + UnicastSource tcpip.Address + MulticastDestination tcpip.Address +} + +// InstalledRoute represents a route that is in the installed state. +// +// If a route is in the installed state, then it may be used to forward +// multicast packets. +type InstalledRoute struct { + expectedInputInterface tcpip.NICID + outgoingInterfaces []OutgoingInterface + // +checkatomic + lastUsedTimestamp atomicbitops.Int64 +} + +// ExpectedInputInterface returns the expected input interface for the route. +func (r *InstalledRoute) ExpectedInputInterface() tcpip.NICID { + return r.expectedInputInterface +} + +// OutgoingInterfaces returns the outgoing interfaces for the route. +func (r *InstalledRoute) OutgoingInterfaces() []OutgoingInterface { + return r.outgoingInterfaces +} + +// LastUsedTimestamp returns a Unix based timestamp in microseconds that +// corresponds to the last time the route was used or updated. +func (r *InstalledRoute) LastUsedTimestamp() int64 { + return r.lastUsedTimestamp.Load() +} + +// SetLastUsedTimestamp sets the time that the route was last used. +// +// Callers should invoke this anytime the route is used to forward a packet. +func (r *InstalledRoute) SetLastUsedTimestamp(time time.Time) { + r.lastUsedTimestamp.Store(time.UnixMicro()) +} + +// OutgoingInterface represents an interface that packets should be forwarded +// out of. +type OutgoingInterface struct { + // ID corresponds to the outgoing NIC. + ID tcpip.NICID + + // MinTTL represents the minumum TTL/HopLimit a multicast packet must have to + // be sent through the outgoing interface. + MinTTL uint8 +} + +// pendingRoute represents a route that is in the "pending" state. +// +// A route is in the pending state if an installed route does not yet exist +// for the entry. For such routes, packets are added to an expiring queue until +// a route is installed. +type pendingRoute struct { + packets []*stack.PacketBuffer +} + +func newPendingRoute(maxSize uint8) pendingRoute { + return pendingRoute{packets: make([]*stack.PacketBuffer, 0, maxSize)} +} + +// Dequeue removes the first element in the queue and returns it. +// +// If the queue is empty, then an error will be returned. +func (p *pendingRoute) Dequeue() (*stack.PacketBuffer, error) { + if len(p.packets) == 0 { + return nil, errors.New("dequeue called on queue empty") + } + val := p.packets[0] + p.packets[0] = nil + p.packets = p.packets[1:] + return val, nil +} + +// IsEmpty returns true if the queue contains no more elements. Otherwise, +// returns false. +func (p *pendingRoute) IsEmpty() bool { + return len(p.packets) == 0 +} + +// DefaultMaxPendingQueueSize corresponds to the number of elements that can be +// in the packet queue for a pending route. +// +// Matches the Linux default queue size: +// https://github.com/torvalds/linux/blob/26291c54e11/net/ipv6/ip6mr.c#L1186 +const DefaultMaxPendingQueueSize uint8 = 3 + +// Config represents the options for configuring a RouteTable. +type Config struct { + // MaxPendingQueueSize corresponds to the maximum number of queued packets + // for a pending route. + // + // If the caller attempts to queue a packet and the queue already contains + // MaxPendingQueueSize elements, then the packet will be rejected and should + // not be forwarded. + MaxPendingQueueSize uint8 + + // Clock represents the clock that should be used to obtain the current time. + // + // This field is required and must have a non-nil value. + Clock tcpip.Clock +} + +// DefaultConfig returns the default configuration for the table. +func DefaultConfig(clock tcpip.Clock) Config { + return Config{MaxPendingQueueSize: DefaultMaxPendingQueueSize, Clock: clock} +} + +// Init initializes the RouteTable with the provided config. +// +// An error is returned if the config is not valid. +// +// Must be called before any other function on the table. +func (r *RouteTable) Init(config Config) error { + r.installedMu.Lock() + defer r.installedMu.Unlock() + r.pendingMu.Lock() + defer r.pendingMu.Unlock() + + if r.installedRoutes != nil { + return ErrAlreadyInitialized + } + + if config.Clock == nil { + return ErrMissingClock + } + + r.config = config + r.installedRoutes = make(map[RouteKey]*InstalledRoute) + r.pendingRoutes = make(map[RouteKey]pendingRoute) + return nil +} + +// NewInstalledRoute instatiates an installed route for the table. +func (r *RouteTable) NewInstalledRoute(inputInterface tcpip.NICID, outgoingInterfaces []OutgoingInterface) *InstalledRoute { + return &InstalledRoute{ + expectedInputInterface: inputInterface, + outgoingInterfaces: outgoingInterfaces, + lastUsedTimestamp: atomicbitops.FromInt64(r.config.Clock.Now().UnixMicro()), + } +} + +// GetRouteResult represents the result of calling +// RouteTable.GetRouteOrInsertPending. +type GetRouteResult struct { + // PendingRouteState represents the observed state of any applicable + // PendingRoute. + PendingRouteState + + // InstalledRoute represents the existing installed route. This field will + // only be populated if the PendingRouteState is PendingRouteStateNone. + *InstalledRoute +} + +// PendingRouteState represents the state of a PendingRoute as observed by the +// RouteTable.GetRouteOrInsertPending method. +type PendingRouteState uint8 + +const ( + // PendingRouteStateNone indicates that no pending route exists. In such a + // case, the GetRouteResult will contain an InstalledRoute. + PendingRouteStateNone PendingRouteState = iota + + // PendingRouteStateAppended indicates that the packet was queued in an + // existing pending route. + PendingRouteStateAppended + + // PendingRouteStateInstalled indicates that a pending route was newly + // inserted into the RouteTable. In such a case, callers should typically + // emit a missing route event. + PendingRouteStateInstalled +) + +func (e PendingRouteState) String() string { + switch e { + case PendingRouteStateNone: + return "PendingRouteStateNone" + case PendingRouteStateAppended: + return "PendingRouteStateAppended" + case PendingRouteStateInstalled: + return "PendingRouteStateInstalled" + default: + return fmt.Sprintf("%d", uint8(e)) + } +} + +// GetRouteOrInsertPending attempts to fetch the installed route that matches +// the provided key. +// +// If no matching installed route is found, then the pkt is queued in a +// pending route. The GetRouteResult.PendingRouteState will indicate whether +// the pkt was queued in a new pending route or an existing one. +// +// If the relevant pending route queue is at max capacity, then +// ErrNoBufferSpace is returned. In such a case, callers are typically expected +// to only deliver the pkt locally (if relevant). +func (r *RouteTable) GetRouteOrInsertPending(key RouteKey, pkt *stack.PacketBuffer) (GetRouteResult, error) { + r.installedMu.RLock() + defer r.installedMu.RUnlock() + + if route, ok := r.installedRoutes[key]; ok { + return GetRouteResult{PendingRouteState: PendingRouteStateNone, InstalledRoute: route}, nil + } + + r.pendingMu.Lock() + defer r.pendingMu.Unlock() + + pendingRoute, pendingRouteState := r.getOrCreatePendingRouteRLocked(key) + if len(pendingRoute.packets) >= int(r.config.MaxPendingQueueSize) { + // The incoming packet is rejected if the pending queue is already at max + // capacity. This behavior matches the Linux implementation: + // https://github.com/torvalds/linux/blob/ae085d7f936/net/ipv4/ipmr.c#L1147 + return GetRouteResult{}, ErrNoBufferSpace + } + pendingRoute.packets = append(pendingRoute.packets, pkt) + r.pendingRoutes[key] = pendingRoute + + return GetRouteResult{PendingRouteState: pendingRouteState, InstalledRoute: nil}, nil +} + +// +checklocks:r.pendingMu +func (r *RouteTable) getOrCreatePendingRouteRLocked(key RouteKey) (pendingRoute, PendingRouteState) { + if pendingRoute, ok := r.pendingRoutes[key]; ok { + return pendingRoute, PendingRouteStateAppended + } + + pendingRoute := newPendingRoute(r.config.MaxPendingQueueSize) + return pendingRoute, PendingRouteStateInstalled +} diff --git a/pkg/tcpip/network/internal/multicast/route_table_test.go b/pkg/tcpip/network/internal/multicast/route_table_test.go new file mode 100644 index 000000000..9bca1177a --- /dev/null +++ b/pkg/tcpip/network/internal/multicast/route_table_test.go @@ -0,0 +1,186 @@ +// Copyright 2022 The gVisor Authors. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package multicast + +import ( + "os" + "testing" + "time" + + "github.com/google/go-cmp/cmp" + "github.com/google/go-cmp/cmp/cmpopts" + "gvisor.dev/gvisor/pkg/atomicbitops" + "gvisor.dev/gvisor/pkg/refs" + "gvisor.dev/gvisor/pkg/refsvfs2" + "gvisor.dev/gvisor/pkg/tcpip" + "gvisor.dev/gvisor/pkg/tcpip/buffer" + "gvisor.dev/gvisor/pkg/tcpip/faketime" + "gvisor.dev/gvisor/pkg/tcpip/stack" + "gvisor.dev/gvisor/pkg/tcpip/testutil" +) + +const ( + defaultMinTTL = 10 + inputNICID tcpip.NICID = 1 + outgoingNICID tcpip.NICID = 2 +) + +var ( + defaultAddress = testutil.MustParse4("192.168.1.1") + defaultRouteKey = RouteKey{UnicastSource: defaultAddress, MulticastDestination: defaultAddress} + defaultOutgoingInterfaces = []OutgoingInterface{{ID: outgoingNICID, MinTTL: defaultMinTTL}} +) + +func newPacketBuffer(body string) *stack.PacketBuffer { + return stack.NewPacketBuffer(stack.PacketBufferOptions{ + Data: buffer.View(body).ToVectorisedView(), + }) +} + +type configOption func(*Config) + +func withMaxPendingQueueSize(size uint8) configOption { + return func(c *Config) { + c.MaxPendingQueueSize = size + } +} + +func withClock(clock tcpip.Clock) configOption { + return func(c *Config) { + c.Clock = clock + } +} + +func defaultConfig(opts ...configOption) Config { + c := &Config{ + MaxPendingQueueSize: DefaultMaxPendingQueueSize, + Clock: faketime.NewManualClock(), + } + + for _, opt := range opts { + opt(c) + } + + return *c +} + +func TestInit(t *testing.T) { + tests := []struct { + name string + config Config + invokeTwice bool + wantErr error + }{ + { + name: "MissingClock", + config: defaultConfig(withClock(nil)), + invokeTwice: false, + wantErr: ErrMissingClock, + }, + { + name: "AlreadyInitialized", + config: defaultConfig(), + invokeTwice: true, + wantErr: ErrAlreadyInitialized, + }, + { + name: "ValidConfig", + config: defaultConfig(), + invokeTwice: false, + wantErr: nil, + }, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + table := RouteTable{} + err := table.Init(tc.config) + + if tc.invokeTwice { + err = table.Init(tc.config) + } + + if !cmp.Equal(err, tc.wantErr, cmpopts.EquateErrors()) { + t.Errorf("got table.Init(%#v) = %s, want %s", tc.config, err, tc.wantErr) + } + }) + } +} + +func TestNewInstalledRoute(t *testing.T) { + table := RouteTable{} + clock := faketime.NewManualClock() + clock.Advance(5 * time.Second) + + config := defaultConfig(withClock(clock)) + if err := table.Init(config); err != nil { + t.Fatalf("table.Init(%#v): %s", config, err) + } + + route := table.NewInstalledRoute(inputNICID, defaultOutgoingInterfaces) + expectedRoute := &InstalledRoute{expectedInputInterface: inputNICID, outgoingInterfaces: defaultOutgoingInterfaces, lastUsedTimestamp: atomicbitops.FromInt64(clock.Now().UnixMicro())} + + if diff := cmp.Diff(expectedRoute, route, cmp.Comparer(func(a *InstalledRoute, b *InstalledRoute) bool { + if !cmp.Equal(a.OutgoingInterfaces(), b.OutgoingInterfaces()) { + return false + } + + if a.ExpectedInputInterface() != b.ExpectedInputInterface() { + return false + } + + return a.LastUsedTimestamp() == b.LastUsedTimestamp() + })); diff != "" { + t.Errorf("installed route mismatch (-want +got):\n%s", diff) + } +} + +func TestPendingRouteStates(t *testing.T) { + table := RouteTable{} + config := defaultConfig(withMaxPendingQueueSize(2)) + if err := table.Init(config); err != nil { + t.Fatalf("table.Init(%#v): %s", config, err) + } + + pkt := newPacketBuffer("hello") + defer pkt.DecRef() + // Queue two pending packets for the same route. The PendingRouteState should + // transition from PendingRouteStateInstalled to PendingRouteStateAppended. + for _, wantPendingRouteState := range [...]PendingRouteState{PendingRouteStateInstalled, PendingRouteStateAppended} { + routeResult, err := table.GetRouteOrInsertPending(defaultRouteKey, pkt) + + if err != nil { + t.Errorf("got table.GetRouteOrInsertPending(%#v, %#v) = (_, %v), want = (_, nil)", defaultRouteKey, pkt, err) + } + + expectedResult := GetRouteResult{PendingRouteState: wantPendingRouteState} + if diff := cmp.Diff(expectedResult, routeResult); diff != "" { + t.Errorf("table.GetRouteOrInsertPending(%#v, %#v) GetRouteResult mismatch (-want +got):\n%s", defaultRouteKey, pkt, diff) + } + } + + // Queuing a third packet should yield an error since the pending queue is + // already at max capacity. + if _, err := table.GetRouteOrInsertPending(defaultRouteKey, pkt); err != ErrNoBufferSpace { + t.Errorf("got table.GetRouteOrInsertPending(%#v, %#v) = (_, %v), want = (_, ErrNoBufferSpace)", defaultRouteKey, pkt, err) + } +} + +func TestMain(m *testing.M) { + refs.SetLeakMode(refs.LeaksPanic) + code := m.Run() + refsvfs2.DoLeakCheck() + os.Exit(code) +}