diff --git a/dataplane/saiserver/routing.go b/dataplane/saiserver/routing.go index 1e66c74e..3c50e5b5 100644 --- a/dataplane/saiserver/routing.go +++ b/dataplane/saiserver/routing.go @@ -1280,7 +1280,7 @@ type vlanMember struct { } type vlan struct { - mu sync.Mutex + mu sync.RWMutex saipb.UnimplementedVlanServer mgr *attrmgr.AttrMgr dataplane switchDataplaneAPI @@ -1300,6 +1300,8 @@ func newVlan(mgr *attrmgr.AttrMgr, dataplane switchDataplaneAPI, s *grpc.Server) } func (vlan *vlan) vidByOid(oid uint64) (uint32, error) { + vlan.mu.RLock() + defer vlan.mu.RUnlock() for v, o := range vlan.oidByVId { if o == oid { return v, nil @@ -1309,6 +1311,8 @@ func (vlan *vlan) vidByOid(oid uint64) (uint32, error) { } func (vlan *vlan) memberByOid(oid uint64) *vlanMember { + vlan.mu.RLock() + defer vlan.mu.RUnlock() for _, v := range vlan.vlans { for mOid, member := range v { if mOid == oid { @@ -1320,8 +1324,8 @@ func (vlan *vlan) memberByOid(oid uint64) *vlanMember { } func (vlan *vlan) memberByPortId(oid uint64) *vlanMember { - vlan.mu.Lock() - defer vlan.mu.Unlock() + vlan.mu.RLock() + defer vlan.mu.RUnlock() for _, v := range vlan.vlans { for _, member := range v { if member.PortID == oid { @@ -1332,9 +1336,10 @@ func (vlan *vlan) memberByPortId(oid uint64) *vlanMember { return nil } +// oidByVid returns the VLAN OID for the given VLAN ID. func (vlan *vlan) oidByVid(vid uint32) (uint64, bool) { - vlan.mu.Lock() - defer vlan.mu.Unlock() + vlan.mu.RLock() + defer vlan.mu.RUnlock() oid, ok := vlan.oidByVId[vid] return oid, ok } @@ -1368,9 +1373,6 @@ func (vlan *vlan) CreateVlan(ctx context.Context, r *saipb.CreateVlanRequest) (* TamObject: []uint64{}, } vlan.mgr.StoreAttributes(id, attrs) - vlan.vlans[id] = map[uint64]*vlanMember{} - vlan.oidByVId[r.GetVlanId()] = id - floodID := fmt.Sprintf("vlan_flood_%d", r.GetVlanId()) if _, err := vlan.dataplane.PortCreate(ctx, &fwdpb.PortCreateRequest{ ContextId: &fwdpb.ContextId{Id: vlan.dataplane.ID()}, @@ -1388,6 +1390,14 @@ func (vlan *vlan) CreateVlan(ctx context.Context, r *saipb.CreateVlanRequest) (* return nil, err } + vlan.mu.Lock() + if _, ok := vlan.oidByVId[r.GetVlanId()]; ok { + vlan.mu.Unlock() + return nil, fmt.Errorf("found existing VLAN %d", r.GetVlanId()) + } + vlan.vlans[id] = map[uint64]*vlanMember{} + vlan.oidByVId[r.GetVlanId()] = id + vlan.mu.Unlock() return &saipb.CreateVlanResponse{ Oid: id, }, nil @@ -1530,7 +1540,12 @@ func (vlan *vlan) CreateVlanMember(ctx context.Context, r *saipb.CreateVlanMembe vlanAttrResp.GetAttr().MemberList = append(vlanAttrResp.GetAttr().MemberList, mOid) vlan.mgr.StoreAttributes(vOid, vlanAttrResp.GetAttr()) vlan.mu.Lock() - vlan.vlans[vOid][mOid] = &vlanMember{ + members, ok := vlan.vlans[vOid] + if !ok { + vlan.mu.Unlock() + return nil, fmt.Errorf("cannot find VLAN with OID %d", vOid) + } + members[mOid] = &vlanMember{ Oid: mOid, PortID: portID, Vid: vId, @@ -1541,7 +1556,10 @@ func (vlan *vlan) CreateVlanMember(ctx context.Context, r *saipb.CreateVlanMembe // Fetch the original vlan from the old vlan member and remove the member from that vlan if member != nil { - preVlanOid, _ := vlan.oidByVid(member.Vid) + preVlanOid, ok := vlan.oidByVid(member.Vid) + if !ok { + return nil, fmt.Errorf("cannot find VLAN OID for VID %d", member.Vid) + } vlanAttrReq = &saipb.GetVlanAttributeRequest{Oid: preVlanOid, AttrType: []saipb.VlanAttr{saipb.VlanAttr_VLAN_ATTR_MEMBER_LIST}} vlanAttrResp = &saipb.GetVlanAttributeResponse{} if err := vlan.mgr.PopulateAttributes(vlanAttrReq, vlanAttrResp); err != nil { @@ -1564,9 +1582,11 @@ func (vlan *vlan) CreateVlanMember(ctx context.Context, r *saipb.CreateVlanMembe } func (vlan *vlan) RemoveVlanMember(ctx context.Context, r *saipb.RemoveVlanMemberRequest) (*saipb.RemoveVlanMemberResponse, error) { + // Claim the member under the lock so that concurrent removals of the same + // OID cannot both proceed, then release the lock before issuing dataplane + // RPCs so vlan.mu is never held across calls into the dataplane. If the + // dataplane removal fails, the claim is undone so the call can be retried. vlan.mu.Lock() - defer vlan.mu.Unlock() - var member *vlanMember var targetVlanOid uint64 found := false @@ -1581,18 +1601,24 @@ func (vlan *vlan) RemoveVlanMember(ctx context.Context, r *saipb.RemoveVlanMembe } if !found { + vlan.mu.Unlock() return nil, fmt.Errorf("cannot find member with OID %d", r.GetOid()) } + delete(vlan.vlans[targetVlanOid], r.GetOid()) + vlan.mu.Unlock() + nid, err := vlan.dataplane.ObjectNID(ctx, &fwdpb.ObjectNIDRequest{ ContextId: &fwdpb.ContextId{Id: vlan.dataplane.ID()}, ObjectId: &fwdpb.ObjectId{Id: fmt.Sprint(member.PortID)}, }) if err != nil { slog.InfoContext(ctx, "Failed to find NID for port", "bridge_port", member.PortID, "err", err) + vlan.restoreMember(targetVlanOid, member) return nil, err } if _, err := vlan.dataplane.TableEntryRemove(ctx, fwdconfig.TableEntryRemoveRequest(vlan.dataplane.ID(), VlanTable).AppendEntry( fwdconfig.EntryDesc(fwdconfig.ExactEntry(fwdconfig.PacketFieldBytes(fwdpb.PacketFieldNum_PACKET_FIELD_NUM_PACKET_PORT_INPUT).WithUint64(nid.GetNid())))).Build()); err != nil { + vlan.restoreMember(targetVlanOid, member) return nil, err } floodID := fmt.Sprintf("vlan_flood_%d", member.Vid) @@ -1603,14 +1629,50 @@ func (vlan *vlan) RemoveVlanMember(ctx context.Context, r *saipb.RemoveVlanMembe AggregateDel: &fwdpb.AggregatePortRemoveMemberUpdateDesc{PortId: &fwdpb.PortId{ObjectId: &fwdpb.ObjectId{Id: fmt.Sprint(member.PortID)}}}, }}, }); err != nil { + vlanReq := fwdconfig.TableEntryAddRequest(vlan.dataplane.ID(), VlanTable).AppendEntry( + fwdconfig.EntryDesc(fwdconfig.ExactEntry(fwdconfig.PacketFieldBytes(fwdpb.PacketFieldNum_PACKET_FIELD_NUM_PACKET_PORT_INPUT).WithUint64(nid.GetNid())))).Build() + if member.Mode != saipb.VlanTaggingMode_VLAN_TAGGING_MODE_TAGGED { + vlanReq.Entries[0].Actions = []*fwdpb.ActionDesc{ + fwdconfig.Action(fwdconfig.EncapAction(fwdpb.PacketHeaderId_PACKET_HEADER_ID_ETHERNET_VLAN)).Build(), + fwdconfig.Action(fwdconfig.UpdateAction(fwdpb.UpdateType_UPDATE_TYPE_SET, fwdpb.PacketFieldNum_PACKET_FIELD_NUM_VLAN_TAG).WithValue(binary.BigEndian.AppendUint16(nil, uint16(member.Vid)))).Build(), + } + } + if _, rerr := vlan.dataplane.TableEntryAdd(ctx, vlanReq); rerr != nil { + slog.WarnContext(ctx, "RemoveVlanMember: failed to re-add VlanTable entry after PortUpdate failure", "port_id", member.PortID, "err", rerr) + } + vlan.restoreMember(targetVlanOid, member) return nil, err } - delete(vlan.vlans[targetVlanOid], r.GetOid()) + // Remove the member OID from the parent VLAN's MEMBER_LIST attribute + // only once the dataplane removal has succeeded. + vlanAttrReq := &saipb.GetVlanAttributeRequest{Oid: targetVlanOid, AttrType: []saipb.VlanAttr{saipb.VlanAttr_VLAN_ATTR_MEMBER_LIST}} + vlanAttrResp := &saipb.GetVlanAttributeResponse{} + if err := vlan.mgr.PopulateAttributes(vlanAttrReq, vlanAttrResp); err != nil { + return nil, err + } + newMemList := []uint64{} + for _, id := range vlanAttrResp.GetAttr().GetMemberList() { + if id != r.GetOid() { + newMemList = append(newMemList, id) + } + } + vlanAttrResp.GetAttr().MemberList = newMemList + vlan.mgr.StoreAttributes(targetVlanOid, vlanAttrResp.GetAttr()) return &saipb.RemoveVlanMemberResponse{}, nil } +// restoreMember re-inserts a member claimed by RemoveVlanMember after its +// dataplane removal failed, unless the parent VLAN has since been removed. +func (vlan *vlan) restoreMember(vlanOid uint64, member *vlanMember) { + vlan.mu.Lock() + defer vlan.mu.Unlock() + if members, ok := vlan.vlans[vlanOid]; ok { + members[member.Oid] = member + } +} + func (vlan *vlan) CreateVlanMembers(ctx context.Context, r *saipb.CreateVlanMembersRequest) (*saipb.CreateVlanMembersResponse, error) { resp := &saipb.CreateVlanMembersResponse{} for _, req := range r.GetReqs() { @@ -1625,6 +1687,8 @@ func (vlan *vlan) CreateVlanMembers(ctx context.Context, r *saipb.CreateVlanMemb func (vlan *vlan) Reset() { slog.Info("resetting vlan") + vlan.mu.Lock() + defer vlan.mu.Unlock() vlan.oidByVId = map[uint32]uint64{} vlan.vlans = map[uint64]map[uint64]*vlanMember{} } diff --git a/dataplane/saiserver/vlan_test.go b/dataplane/saiserver/vlan_test.go index c1f5e3b1..2a2bfda4 100644 --- a/dataplane/saiserver/vlan_test.go +++ b/dataplane/saiserver/vlan_test.go @@ -18,6 +18,8 @@ import ( "context" "encoding/binary" "fmt" + "slices" + "sync" "testing" "google.golang.org/grpc" @@ -233,8 +235,8 @@ func TestVlanOperations(t *testing.T) { if err != nil { t.Fatalf("Failed to get members of the default VLAN: %v", err) } - if len(testm) != 2 && len(defaultm) != 2 { - t.Fatalf("Failed to create a non-default VLAN") + if len(testm) != 2 || len(defaultm) != 2 { + t.Fatalf("Failed to create a non-default VLAN: got testm=%v defaultm=%v", testm, defaultm) } // Remove one port from the non-default VLAN. @@ -252,8 +254,11 @@ func TestVlanOperations(t *testing.T) { if err != nil { t.Fatalf("Failed to get members of default VLAN: %v", err) } - if len(testm) != 1 && len(defaultm) != 2 { - t.Errorf("Failed to remove a non-default VLAN member") + if len(testm) != 1 || len(defaultm) != 2 { + t.Errorf("Failed to remove a non-default VLAN member: got testm=%v defaultm=%v", testm, defaultm) + } + if slices.Contains(testm, testVLANMembers[0]) { + t.Errorf("removed VLAN member %d still in MEMBER_LIST %v", testVLANMembers[0], testm) } // Remove the non-default VLAN. @@ -461,3 +466,237 @@ func TestCreateVlanMemberTaggingMode(t *testing.T) { }) } } + +// lockedVlanDataplane serializes access to the fakeSwitchDataplane methods used +// by the vlan server so that concurrency tests only flag races in the vlan server. +type lockedVlanDataplane struct { + *fakeSwitchDataplane + mu sync.Mutex +} + +func (l *lockedVlanDataplane) PortCreate(ctx context.Context, req *fwdpb.PortCreateRequest) (*fwdpb.PortCreateReply, error) { + l.mu.Lock() + defer l.mu.Unlock() + return l.fakeSwitchDataplane.PortCreate(ctx, req) +} + +func (l *lockedVlanDataplane) PortUpdate(ctx context.Context, req *fwdpb.PortUpdateRequest) (*fwdpb.PortUpdateReply, error) { + l.mu.Lock() + defer l.mu.Unlock() + return l.fakeSwitchDataplane.PortUpdate(ctx, req) +} + +func (l *lockedVlanDataplane) ObjectNID(ctx context.Context, req *fwdpb.ObjectNIDRequest) (*fwdpb.ObjectNIDReply, error) { + l.mu.Lock() + defer l.mu.Unlock() + return l.fakeSwitchDataplane.ObjectNID(ctx, req) +} + +func (l *lockedVlanDataplane) TableEntryAdd(ctx context.Context, req *fwdpb.TableEntryAddRequest) (*fwdpb.TableEntryAddReply, error) { + l.mu.Lock() + defer l.mu.Unlock() + return l.fakeSwitchDataplane.TableEntryAdd(ctx, req) +} + +func (l *lockedVlanDataplane) TableEntryRemove(ctx context.Context, req *fwdpb.TableEntryRemoveRequest) (*fwdpb.TableEntryRemoveReply, error) { + l.mu.Lock() + defer l.mu.Unlock() + return l.fakeSwitchDataplane.TableEntryRemove(ctx, req) +} + +// TestVlanConcurrentAccess exercises the vlan server's internal maps from +// concurrent SAI calls and lookup helpers. It is intended to be run with -race. +func TestVlanConcurrentAccess(t *testing.T) { + const workers = 8 + dplane := &lockedVlanDataplane{fakeSwitchDataplane: &fakeSwitchDataplane{}} + var v *vlan + conn, mgr, stopFn := newTestServer(t, func(mgr *attrmgr.AttrMgr, srv *grpc.Server) { + v = newVlan(mgr, dplane, srv) + }) + defer stopFn() + c := saipb.NewVlanClient(conn) + ctx := context.Background() + + mgr.StoreAttributes(1, &saipb.SwitchAttribute{DefaultStpInstId: proto.Uint64(testStpInstId)}) + for i := 0; i < workers; i++ { + portID := uint64(1000 + i) + mgr.StoreAttributes(portID+10000, &saipb.BridgePortAttribute{PortId: proto.Uint64(portID)}) + } + + done := make(chan struct{}) + var readers sync.WaitGroup + readers.Add(1) + go func() { + defer readers.Done() + for { + select { + case <-done: + return + default: + } + for i := 0; i < workers; i++ { + v.memberByPortId(uint64(1000 + i)) + if oid, ok := v.oidByVid(uint32(100 + i)); ok { + v.vidByOid(oid) + v.memberByOid(oid) + } + } + } + }() + + var writers sync.WaitGroup + errCh := make(chan error, workers) + for i := 0; i < workers; i++ { + writers.Add(1) + go func(i int) { + defer writers.Done() + vResp, err := c.CreateVlan(ctx, &saipb.CreateVlanRequest{Switch: 1, VlanId: proto.Uint32(uint32(100 + i))}) + if err != nil { + errCh <- fmt.Errorf("CreateVlan(%d) failed: %v", 100+i, err) + return + } + mResp, err := c.CreateVlanMember(ctx, &saipb.CreateVlanMemberRequest{ + Switch: 1, + VlanId: proto.Uint64(vResp.GetOid()), + BridgePortId: proto.Uint64(uint64(1000+i) + 10000), + }) + if err != nil { + errCh <- fmt.Errorf("CreateVlanMember(%d) failed: %v", 100+i, err) + return + } + if _, err := c.RemoveVlanMember(ctx, &saipb.RemoveVlanMemberRequest{Oid: mResp.GetOid()}); err != nil { + errCh <- fmt.Errorf("RemoveVlanMember(%d) failed: %v", mResp.GetOid(), err) + return + } + if _, err := c.RemoveVlan(ctx, &saipb.RemoveVlanRequest{Oid: vResp.GetOid()}); err != nil { + errCh <- fmt.Errorf("RemoveVlan(%d) failed: %v", vResp.GetOid(), err) + } + }(i) + } + writers.Wait() + close(done) + readers.Wait() + close(errCh) + for err := range errCh { + t.Error(err) + } + + for i := 0; i < workers; i++ { + if _, ok := v.oidByVid(uint32(100 + i)); ok { + t.Errorf("VLAN %d still present after RemoveVlan", 100+i) + } + if m := v.memberByPortId(uint64(1000 + i)); m != nil { + t.Errorf("port %d still has VLAN member %+v after RemoveVlanMember", 1000+i, m) + } + } +} + +// failingRemoveDataplane fails TableEntryRemove or PortUpdate when instructed. +type failingRemoveDataplane struct { + *fakeSwitchDataplane + failTableRemove bool + failPortUpdate bool +} + +func (f *failingRemoveDataplane) TableEntryAdd(ctx context.Context, req *fwdpb.TableEntryAddRequest) (*fwdpb.TableEntryAddReply, error) { + // If an entry is re-added after a removal (such as during rollback), reset gotEntryRemoveReqs + // so that a subsequent removal is not treated as a double removal. + f.gotEntryRemoveReqs = nil + return f.fakeSwitchDataplane.TableEntryAdd(ctx, req) +} + +func (f *failingRemoveDataplane) TableEntryRemove(ctx context.Context, req *fwdpb.TableEntryRemoveRequest) (*fwdpb.TableEntryRemoveReply, error) { + if f.failTableRemove { + f.failTableRemove = false + return nil, fmt.Errorf("injected TableEntryRemove failure") + } + return f.fakeSwitchDataplane.TableEntryRemove(ctx, req) +} + +func (f *failingRemoveDataplane) PortUpdate(ctx context.Context, req *fwdpb.PortUpdateRequest) (*fwdpb.PortUpdateReply, error) { + if f.failPortUpdate { + f.failPortUpdate = false + return nil, fmt.Errorf("injected PortUpdate failure") + } + return f.fakeSwitchDataplane.PortUpdate(ctx, req) +} + +func TestRemoveVlanMemberRetryAfterDataplaneError(t *testing.T) { + tests := []struct { + name string + setupFail func(d *failingRemoveDataplane) + }{ + { + name: "TableEntryRemove failure", + setupFail: func(d *failingRemoveDataplane) { + d.failTableRemove = true + }, + }, + { + name: "PortUpdate failure", + setupFail: func(d *failingRemoveDataplane) { + d.failPortUpdate = true + }, + }, + } + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + dplane := &failingRemoveDataplane{fakeSwitchDataplane: &fakeSwitchDataplane{}} + var v *vlan + conn, mgr, stopFn := newTestServer(t, func(mgr *attrmgr.AttrMgr, srv *grpc.Server) { + v = newVlan(mgr, dplane, srv) + }) + defer stopFn() + c := saipb.NewVlanClient(conn) + ctx := context.Background() + + mgr.StoreAttributes(1, &saipb.SwitchAttribute{DefaultStpInstId: proto.Uint64(testStpInstId)}) + const bridgePortOid, portID = uint64(100), uint64(10) + mgr.StoreAttributes(bridgePortOid, &saipb.BridgePortAttribute{PortId: proto.Uint64(portID)}) + + vlanResp, err := c.CreateVlan(ctx, &saipb.CreateVlanRequest{Switch: 1, VlanId: proto.Uint32(10)}) + if err != nil { + t.Fatalf("CreateVlan failed: %v", err) + } + memberResp, err := c.CreateVlanMember(ctx, &saipb.CreateVlanMemberRequest{ + Switch: 1, + VlanId: proto.Uint64(vlanResp.GetOid()), + BridgePortId: proto.Uint64(bridgePortOid), + }) + if err != nil { + t.Fatalf("CreateVlanMember failed: %v", err) + } + + getMemberList := func() []uint64 { + t.Helper() + req := &saipb.GetVlanAttributeRequest{Oid: vlanResp.GetOid(), AttrType: []saipb.VlanAttr{saipb.VlanAttr_VLAN_ATTR_MEMBER_LIST}} + resp := &saipb.GetVlanAttributeResponse{} + if err := mgr.PopulateAttributes(req, resp); err != nil { + t.Fatalf("PopulateAttributes(%d) failed: %v", vlanResp.GetOid(), err) + } + return resp.GetAttr().GetMemberList() + } + + tc.setupFail(dplane) + if _, err := c.RemoveVlanMember(ctx, &saipb.RemoveVlanMemberRequest{Oid: memberResp.GetOid()}); err == nil { + t.Fatalf("RemoveVlanMember with failing dataplane succeeded, want error") + } + if m := v.memberByOid(memberResp.GetOid()); m == nil { + t.Fatalf("VLAN member %d forgotten after failed RemoveVlanMember, want it retained for retry", memberResp.GetOid()) + } + if got := getMemberList(); !slices.Contains(got, memberResp.GetOid()) { + t.Errorf("MEMBER_LIST = %v after failed RemoveVlanMember, want it to still contain %d", got, memberResp.GetOid()) + } + + if _, err := c.RemoveVlanMember(ctx, &saipb.RemoveVlanMemberRequest{Oid: memberResp.GetOid()}); err != nil { + t.Fatalf("RemoveVlanMember retry failed: %v", err) + } + if m := v.memberByOid(memberResp.GetOid()); m != nil { + t.Errorf("VLAN member %d still present after successful RemoveVlanMember", memberResp.GetOid()) + } + if got := getMemberList(); slices.Contains(got, memberResp.GetOid()) { + t.Errorf("MEMBER_LIST = %v after successful RemoveVlanMember, want %d removed", got, memberResp.GetOid()) + } + }) + } +}