vikarti.anatra/safing-portmaster
1
0
Fork 0
mirror of https://github.com/safing/portmaster synced 2025-09-04 19:49:15 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 4

Use new call limiter in network state tables

Browse source 
Potentially fixes #1043, #1294
This commit is contained in:
Daniel 2023-09-28 15:04:54 +02:00
parent 6fc7c8c169
commit 577299c95b
5 changed files with 83 additions and 212 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

4
network/state/exists.go
View file

@ -37,7 +37,7 @@ func Exists(pktInfo *packet.Info, now time.Time) (exists bool) {
func (table *tcpTable) exists(pktInfo *packet.Info) (exists bool) { func (table *tcpTable) exists(pktInfo *packet.Info) (exists bool) {
// Update tables if older than the connection that is checked. // Update tables if older than the connection that is checked.
if table.lastUpdateAt.Load() < pktInfo.SeenAt.UnixNano() { if table.lastUpdateAt.Load() < pktInfo.SeenAt.UnixNano() {
table.updateTables(table.updateIter.Load()) table.updateTables()
} }
table.lock.RLock() table.lock.RLock()
@ -64,7 +64,7 @@ func (table *tcpTable) exists(pktInfo *packet.Info) (exists bool) {
func (table *udpTable) exists(pktInfo *packet.Info, now time.Time) (exists bool) { func (table *udpTable) exists(pktInfo *packet.Info, now time.Time) (exists bool) {
// Update tables if older than the connection that is checked. // Update tables if older than the connection that is checked.
if table.lastUpdateAt.Load() < pktInfo.SeenAt.UnixNano() { if table.lastUpdateAt.Load() < pktInfo.SeenAt.UnixNano() {
table.updateTables(table.updateIter.Load()) table.updateTables()
} }
table.lock.RLock() table.lock.RLock()

8
network/state/info.go
View file

@ -25,12 +25,12 @@ type Info struct {
func GetInfo() *Info { func GetInfo() *Info {
info := &Info{} info := &Info{}
info.TCP4Connections, info.TCP4Listeners, _ = tcp4Table.updateTables(tcp4Table.updateIter.Load()) info.TCP4Connections, info.TCP4Listeners = tcp4Table.updateTables()
info.UDP4Binds, _ = udp4Table.updateTables(udp4Table.updateIter.Load()) info.UDP4Binds = udp4Table.updateTables()
if netenv.IPv6Enabled() { if netenv.IPv6Enabled() {
info.TCP6Connections, info.TCP6Listeners, _ = tcp6Table.updateTables(tcp6Table.updateIter.Load()) info.TCP6Connections, info.TCP6Listeners = tcp6Table.updateTables()
info.UDP6Binds, _ = udp6Table.updateTables(udp6Table.updateIter.Load()) info.UDP6Binds = udp6Table.updateTables()
} }
info.UpdateMeta() info.UpdateMeta()

33
network/state/lookup.go
View file

@ -66,20 +66,18 @@ func (table *tcpTable) lookup(pktInfo *packet.Info, fast bool) (
var ( var (
connections []*socket.ConnectionInfo connections []*socket.ConnectionInfo
listeners []*socket.BindInfo listeners []*socket.BindInfo
updateIter uint64
dualStackConnections []*socket.ConnectionInfo dualStackConnections []*socket.ConnectionInfo
dualStackListeners []*socket.BindInfo dualStackListeners []*socket.BindInfo
dualStackUpdateIter uint64
) )
// Search for the socket until found. // Search for the socket until found.
for i := 1; i <= lookupTries; i++ { for i := 1; i <= lookupTries; i++ {
// Get or update tables. // Use existing tables for first check if packet was seen after last table update.
if i == 1 && pktInfo.SeenAt.UnixNano() >= table.lastUpdateAt.Load() { if i == 1 && pktInfo.SeenAt.UnixNano() >= table.lastUpdateAt.Load() {
connections, listeners, updateIter = table.getCurrentTables() connections, listeners = table.getCurrentTables()
} else { } else {
connections, listeners, updateIter = table.updateTables(updateIter) connections, listeners = table.updateTables()
} }
// Check tables for socket. // Check tables for socket.
@ -97,11 +95,11 @@ func (table *tcpTable) lookup(pktInfo *packet.Info, fast bool) (
continue continue
} }
// Get or update tables. // Use existing tables for first check if packet was seen after last table update.
if i == 0 { if i == 1 && pktInfo.SeenAt.UnixNano() >= table.dualStack.lastUpdateAt.Load() {
dualStackConnections, dualStackListeners, dualStackUpdateIter = table.dualStack.getCurrentTables() dualStackConnections, dualStackListeners = table.dualStack.getCurrentTables()
} else { } else {
dualStackConnections, dualStackListeners, dualStackUpdateIter = table.dualStack.updateTables(dualStackUpdateIter) dualStackConnections, dualStackListeners = table.dualStack.updateTables()
} }
// Check tables for socket. // Check tables for socket.
@ -169,20 +167,17 @@ func (table *udpTable) lookup(pktInfo *packet.Info, fast bool) (
// Prepare variables. // Prepare variables.
var ( var (
binds []*socket.BindInfo binds []*socket.BindInfo
updateIter uint64 dualStackBinds []*socket.BindInfo
dualStackBinds []*socket.BindInfo
dualStackUpdateIter uint64
) )
// Search for the socket until found. // Search for the socket until found.
for i := 1; i <= lookupTries; i++ { for i := 1; i <= lookupTries; i++ {
// Get or update tables. // Get or update tables.
if i == 1 && pktInfo.SeenAt.UnixNano() >= table.lastUpdateAt.Load() { if i == 1 && pktInfo.SeenAt.UnixNano() >= table.lastUpdateAt.Load() {
binds, updateIter = table.getCurrentTables() binds = table.getCurrentTables()
} else { } else {
binds, updateIter = table.updateTables(updateIter) binds = table.updateTables()
} }
// Check tables for socket. // Check tables for socket.
@ -212,10 +207,10 @@ func (table *udpTable) lookup(pktInfo *packet.Info, fast bool) (
} }
// Get or update tables. // Get or update tables.
if i == 0 { if i == 1 && pktInfo.SeenAt.UnixNano() >= table.lastUpdateAt.Load() {
dualStackBinds, dualStackUpdateIter = table.dualStack.getCurrentTables() dualStackBinds = table.dualStack.getCurrentTables()
} else { } else {
dualStackBinds, dualStackUpdateIter = table.dualStack.updateTables(dualStackUpdateIter) dualStackBinds = table.dualStack.updateTables()
} }
// Check tables for socket. // Check tables for socket.

121
network/state/tcp.go
View file

@ -7,10 +7,13 @@ import (
"time" "time"
"github.com/safing/portbase/log" "github.com/safing/portbase/log"
"github.com/safing/portbase/utils"
"github.com/safing/portmaster/network/socket" "github.com/safing/portmaster/network/socket"
) )
const maxUpdateTries = 100 const (
minDurationBetweenTableUpdates = 10 * time.Millisecond
)
type tcpTable struct { type tcpTable struct {
version int version int
@ -19,29 +22,26 @@ type tcpTable struct {
listeners []*socket.BindInfo listeners []*socket.BindInfo
lock sync.RWMutex lock sync.RWMutex
updateIter atomic.Uint64
// lastUpdateAt stores the time when the tables where last updated as unix nanoseconds. // lastUpdateAt stores the time when the tables where last updated as unix nanoseconds.
lastUpdateAt atomic.Int64 lastUpdateAt atomic.Int64
fetchingLock sync.Mutex fetchLimiter *utils.CallLimiter
fetchingInProgress bool fetchTable func() (connections []*socket.ConnectionInfo, listeners []*socket.BindInfo, err error)
fetchingDoneSignal chan struct{}
fetchTable func() (connections []*socket.ConnectionInfo, listeners []*socket.BindInfo, err error)
dualStack *tcpTable dualStack *tcpTable
} }
var ( var (
tcp6Table = &tcpTable{ tcp6Table = &tcpTable{
version: 6, version: 6,
fetchingDoneSignal: make(chan struct{}), fetchLimiter: utils.NewCallLimiter(minDurationBetweenTableUpdates),
fetchTable: getTCP6Table, fetchTable: getTCP6Table,
} }
tcp4Table = &tcpTable{ tcp4Table = &tcpTable{
version: 4, version: 4,
fetchingDoneSignal: make(chan struct{}), fetchLimiter: utils.NewCallLimiter(minDurationBetweenTableUpdates),
fetchTable: getTCP4Table, fetchTable: getTCP4Table,
} }
) )
@ -54,97 +54,38 @@ func EnableTCPDualStack() {
func (table *tcpTable) getCurrentTables() ( func (table *tcpTable) getCurrentTables() (
connections []*socket.ConnectionInfo, connections []*socket.ConnectionInfo,
listeners []*socket.BindInfo, listeners []*socket.BindInfo,
updateIter uint64,
) { ) {
table.lock.RLock() table.lock.RLock()
defer table.lock.RUnlock() defer table.lock.RUnlock()
return table.connections, table.listeners, table.updateIter.Load() return table.connections, table.listeners
} }
func (table *tcpTable) checkFetchingState() (fetch bool, signal chan struct{}) { func (table *tcpTable) updateTables() (
table.fetchingLock.Lock()
defer table.fetchingLock.Unlock()
// If fetching is already in progress, just return the signal.
if table.fetchingInProgress {
return false, table.fetchingDoneSignal
}
// Otherwise, tell caller to fetch.
table.fetchingInProgress = true
return true, nil
}
func (table *tcpTable) signalFetchComplete() {
table.fetchingLock.Lock()
defer table.fetchingLock.Unlock()
// Set fetching state.
table.fetchingInProgress = false
// Signal waiting goroutines.
close(table.fetchingDoneSignal)
table.fetchingDoneSignal = make(chan struct{})
}
func (table *tcpTable) updateTables(previousUpdateIter uint64) (
connections []*socket.ConnectionInfo, connections []*socket.ConnectionInfo,
listeners []*socket.BindInfo, listeners []*socket.BindInfo,
updateIter uint64,
) { ) {
var tries int // Fetch tables.
table.fetchLimiter.Do(func() {
// Attempt to update the tables until we get a new version of the tables. // Fetch new tables from system.
for previousUpdateIter == table.updateIter.Load() { connections, listeners, err := table.fetchTable()
// Abort if it takes too long. if err != nil {
tries++ log.Warningf("state: failed to get TCP%d socket table: %s", table.version, err)
if tries > maxUpdateTries { return
log.Warningf("state: failed to upate TCP%d socket table %d times", table.version, tries-1)
return table.getCurrentTables()
} }
// Check if someone is fetching or if we should fetch. // Pre-check for any listeners.
fetch, signal := table.checkFetchingState() for _, bindInfo := range listeners {
if fetch { bindInfo.ListensAny = bindInfo.Local.IP.Equal(net.IPv4zero) || bindInfo.Local.IP.Equal(net.IPv6zero)
defer table.signalFetchComplete()
// Just to be sure, check again if there is a new version.
if previousUpdateIter < table.updateIter.Load() {
return table.getCurrentTables()
}
// Wait for 5 milliseconds.
time.Sleep(5 * time.Millisecond)
// Fetch new tables from system.
connections, listeners, err := table.fetchTable()
if err != nil {
log.Warningf("state: failed to get TCP%d socket table: %s", table.version, err)
// Return the current tables as fallback, as we need to trigger the defer to complete the fetch.
return table.getCurrentTables()
}
// Pre-check for any listeners.
for _, bindInfo := range listeners {
bindInfo.ListensAny = bindInfo.Local.IP.Equal(net.IPv4zero) || bindInfo.Local.IP.Equal(net.IPv6zero)
}
// Apply new tables.
table.lock.Lock()
defer table.lock.Unlock()
table.connections = connections
table.listeners = listeners
table.updateIter.Add(1)
table.lastUpdateAt.Store(time.Now().UnixNano())
// Return new tables immediately.
return table.connections, table.listeners, table.updateIter.Load()
} }
// Otherwise, wait for fetch to complete. // Apply new tables.
<-signal table.lock.Lock()
} defer table.lock.Unlock()
table.connections = connections
table.listeners = listeners
table.lastUpdateAt.Store(time.Now().UnixNano())
})
return table.getCurrentTables() return table.getCurrentTables()
} }

129
network/state/udp.go
View file

@ -9,6 +9,7 @@ import (
"time" "time"
"github.com/safing/portbase/log" "github.com/safing/portbase/log"
"github.com/safing/portbase/utils"
"github.com/safing/portmaster/netenv" "github.com/safing/portmaster/netenv"
"github.com/safing/portmaster/network/packet" "github.com/safing/portmaster/network/packet"
"github.com/safing/portmaster/network/socket" "github.com/safing/portmaster/network/socket"
@ -20,14 +21,11 @@ type udpTable struct {
binds []*socket.BindInfo binds []*socket.BindInfo
lock sync.RWMutex lock sync.RWMutex
updateIter atomic.Uint64
// lastUpdateAt stores the time when the tables where last updated as unix nanoseconds. // lastUpdateAt stores the time when the tables where last updated as unix nanoseconds.
lastUpdateAt atomic.Int64 lastUpdateAt atomic.Int64
fetchingLock sync.Mutex fetchLimiter *utils.CallLimiter
fetchingInProgress bool fetchTable func() (binds []*socket.BindInfo, err error)
fetchingDoneSignal chan struct{}
fetchTable func() (binds []*socket.BindInfo, err error)
states map[string]map[string]*udpState states map[string]map[string]*udpState
statesLock sync.Mutex statesLock sync.Mutex
@ -53,17 +51,17 @@ const (
var ( var (
udp6Table = &udpTable{ udp6Table = &udpTable{
version: 6, version: 6,
fetchingDoneSignal: make(chan struct{}), fetchLimiter: utils.NewCallLimiter(minDurationBetweenTableUpdates),
fetchTable: getUDP6Table, fetchTable: getUDP6Table,
states: make(map[string]map[string]*udpState), states: make(map[string]map[string]*udpState),
} }
udp4Table = &udpTable{ udp4Table = &udpTable{
version: 4, version: 4,
fetchingDoneSignal: make(chan struct{}), fetchLimiter: utils.NewCallLimiter(minDurationBetweenTableUpdates),
fetchTable: getUDP4Table, fetchTable: getUDP4Table,
states: make(map[string]map[string]*udpState), states: make(map[string]map[string]*udpState),
} }
) )
@ -73,97 +71,34 @@ func EnableUDPDualStack() {
udp4Table.dualStack = udp6Table udp4Table.dualStack = udp6Table
} }
func (table *udpTable) getCurrentTables() ( func (table *udpTable) getCurrentTables() (binds []*socket.BindInfo) {
binds []*socket.BindInfo,
updateIter uint64,
) {
table.lock.RLock() table.lock.RLock()
defer table.lock.RUnlock() defer table.lock.RUnlock()
return table.binds, table.updateIter.Load() return table.binds
} }
func (table *udpTable) checkFetchingState() (fetch bool, signal chan struct{}) { func (table *udpTable) updateTables() (binds []*socket.BindInfo) {
table.fetchingLock.Lock() // Fetch tables.
defer table.fetchingLock.Unlock() table.fetchLimiter.Do(func() {
// Fetch new tables from system.
// If fetching is already in progress, just return the signal. binds, err := table.fetchTable()
if table.fetchingInProgress { if err != nil {
return false, table.fetchingDoneSignal log.Warningf("state: failed to get UDP%d socket table: %s", table.version, err)
} return
// Otherwise, tell caller to fetch.
table.fetchingInProgress = true
return true, nil
}
func (table *udpTable) signalFetchComplete() {
table.fetchingLock.Lock()
defer table.fetchingLock.Unlock()
// Set fetching state.
table.fetchingInProgress = false
// Signal waiting goroutines.
close(table.fetchingDoneSignal)
table.fetchingDoneSignal = make(chan struct{})
}
func (table *udpTable) updateTables(previousUpdateIter uint64) (
binds []*socket.BindInfo,
updateIter uint64,
) {
var tries int
// Attempt to update the tables until we get a new version of the tables.
for previousUpdateIter == table.updateIter.Load() {
// Abort if it takes too long.
tries++
if tries > maxUpdateTries {
log.Warningf("state: failed to upate UDP%d socket table %d times", table.version, tries-1)
return table.getCurrentTables()
} }
// Check if someone is fetching or if we should fetch. // Pre-check for any listeners.
fetch, signal := table.checkFetchingState() for _, bindInfo := range binds {
if fetch { bindInfo.ListensAny = bindInfo.Local.IP.Equal(net.IPv4zero) || bindInfo.Local.IP.Equal(net.IPv6zero)
defer table.signalFetchComplete()
// Just to be sure, check again if there is a new version.
if previousUpdateIter < table.updateIter.Load() {
return table.getCurrentTables()
}
// Wait for 5 milliseconds.
time.Sleep(5 * time.Millisecond)
// Fetch new tables from system.
binds, err := table.fetchTable()
if err != nil {
log.Warningf("state: failed to get UDP%d socket table: %s", table.version, err)
// Return the current tables as fallback, as we need to trigger the defer to complete the fetch.
return table.getCurrentTables()
}
// Pre-check for any listeners.
for _, bindInfo := range binds {
bindInfo.ListensAny = bindInfo.Local.IP.Equal(net.IPv4zero) || bindInfo.Local.IP.Equal(net.IPv6zero)
}
// Apply new tables.
table.lock.Lock()
defer table.lock.Unlock()
table.binds = binds
table.updateIter.Add(1)
table.lastUpdateAt.Store(time.Now().UnixNano())
// Return new tables immediately.
return table.binds, table.updateIter.Load()
} }
// Otherwise, wait for fetch to complete. // Apply new tables.
<-signal table.lock.Lock()
} defer table.lock.Unlock()
table.binds = binds
table.lastUpdateAt.Store(time.Now().UnixNano())
})
return table.getCurrentTables() return table.getCurrentTables()
} }
@ -172,11 +107,11 @@ func (table *udpTable) updateTables(previousUpdateIter uint64) (
func CleanUDPStates(_ context.Context) { func CleanUDPStates(_ context.Context) {
now := time.Now().UTC() now := time.Now().UTC()
udp4Table.updateTables(udp4Table.updateIter.Load()) udp4Table.updateTables()
udp4Table.cleanStates(now) udp4Table.cleanStates(now)
if netenv.IPv6Enabled() { if netenv.IPv6Enabled() {
udp6Table.updateTables(udp6Table.updateIter.Load()) udp6Table.updateTables()
udp6Table.cleanStates(now) udp6Table.cleanStates(now)
} }
} }