return err
}
-/* Must hold device and net lock
- */
-func unsafeUpdateBind(device *Device) error {
+func (device *Device) BindUpdate() error {
+ device.mutex.Lock()
+ defer device.mutex.Unlock()
+
+ netc := &device.net
+ netc.mutex.Lock()
+ defer netc.mutex.Unlock()
// close existing sockets
return err
}
- // assumption: netc.update WaitGroup should be exactly 1
-
// open new sockets
if device.isUp.Get() {
- device.log.Debug.Println("UDP bind updating")
-
// bind to new port
var err error
- netc := &device.net
netc.bind, netc.port, err = CreateBind(netc.port)
if err != nil {
netc.bind = nil
peer.mutex.Unlock()
}
- // decrease waitgroup to 0
+ // start receiving routines
go device.RoutineReceiveIncoming(ipv4.Version, netc.bind)
go device.RoutineReceiveIncoming(ipv6.Version, netc.bind)
return nil
}
-func closeBind(device *Device) error {
+func (device *Device) BindClose() error {
device.mutex.Lock()
device.net.mutex.Lock()
err := unsafeCloseBind(device)
)
type Device struct {
- isUp AtomicBool // device is up (TUN interface up)?
+ isUp AtomicBool // device is (going) up
isClosed AtomicBool // device is closed? (acting as guard)
log *Logger // collection of loggers for levels
idCounter uint // for assigning debug ids to peers
device TUNDevice
mtu int32
}
+ state struct {
+ mutex deadlock.Mutex
+ changing AtomicBool
+ current bool
+ }
pool struct {
messageBuffers sync.Pool
}
mac CookieChecker
}
-func (device *Device) Up() {
- device.mutex.Lock()
- defer device.mutex.Unlock()
+func deviceUpdateState(device *Device) {
- device.net.mutex.Lock()
- defer device.net.mutex.Unlock()
+ // check if state already being updated (guard)
- if device.isUp.Swap(true) {
+ if device.state.changing.Swap(true) {
return
}
- unsafeUpdateBind(device)
+ // compare to current state of device
+
+ device.state.mutex.Lock()
+
+ newIsUp := device.isUp.Get()
+
+ if newIsUp == device.state.current {
+ device.state.mutex.Unlock()
+ device.state.changing.Set(false)
+ return
+ }
+
+ device.state.mutex.Unlock()
+
+ // change state of device
+
+ switch newIsUp {
+ case true:
+
+ // start listener
+
+ if err := device.BindUpdate(); err != nil {
+ device.isUp.Set(false)
+ break
+ }
+
+ // start every peer
+
+ for _, peer := range device.peers {
+ peer.Start()
+ }
+
+ case false:
+
+ // stop listening
+
+ device.BindClose()
- for _, peer := range device.peers {
- peer.Start()
+ // stop every peer
+
+ for _, peer := range device.peers {
+ peer.Stop()
+ }
}
+
+ // update state variables
+ // and check for state change in the mean time
+
+ device.state.current = newIsUp
+ device.state.changing.Set(false)
+ deviceUpdateState(device)
}
-func (device *Device) Down() {
- device.mutex.Lock()
- defer device.mutex.Unlock()
+func (device *Device) Up() {
+
+ // closed device cannot be brought up
- if !device.isUp.Swap(false) {
+ if device.isClosed.Get() {
return
}
- closeBind(device)
+ device.state.mutex.Lock()
+ device.isUp.Set(true)
+ device.state.mutex.Unlock()
+ deviceUpdateState(device)
+}
- for _, peer := range device.peers {
- peer.Stop()
- }
+func (device *Device) Down() {
+ device.state.mutex.Lock()
+ device.isUp.Set(false)
+ device.state.mutex.Unlock()
+ deviceUpdateState(device)
}
/* Warning:
if !ok {
return
}
- peer.Stop()
device.routingTable.RemovePeer(peer)
delete(device.peers, key)
}
func (device *Device) RemoveAllPeers() {
device.mutex.Lock()
defer device.mutex.Unlock()
- for key := range device.peers {
- removePeerUnsafe(device, key)
+
+ for key, peer := range device.peers {
+ peer.Stop()
+ peer, ok := device.peers[key]
+ if !ok {
+ return
+ }
+ device.routingTable.RemovePeer(peer)
+ delete(device.peers, key)
}
}
func (device *Device) Close() {
+ device.log.Info.Println("Device closing")
if device.isClosed.Swap(true) {
return
}
- device.log.Info.Println("Closing device")
- device.RemoveAllPeers()
device.signal.stop.Broadcast()
device.tun.device.Close()
- closeBind(device)
+ device.BindClose()
+ device.isUp.Set(false)
+ println("remove")
+ device.RemoveAllPeers()
+ device.log.Info.Println("Interface closed")
}
func (device *Device) Wait() chan struct{} {
"encoding/base64"
"errors"
"fmt"
+ "github.com/sasha-s/go-deadlock"
"sync"
"time"
)
type Peer struct {
id uint
- mutex sync.RWMutex
+ isRunning AtomicBool
+ mutex deadlock.RWMutex
persistentKeepaliveInterval uint64
keyPairs KeyPairs
handshake Handshake
lastHandshakeNano int64 // nano seconds since epoch
}
time struct {
- mutex sync.RWMutex
+ mutex deadlock.RWMutex
lastSend time.Time // last send message
lastHandshake time.Time // last completed handshake
nextKeepalive time.Time
inbound chan *QueueInboundElement // sequential ordering of work
}
routines struct {
- mutex sync.Mutex // held when stopping / starting routines
+ mutex deadlock.Mutex // held when stopping / starting routines
starting sync.WaitGroup // routines pending start
stopping sync.WaitGroup // routines pending stop
stop Signal // size 0, stop all goroutines in peer
}
func (device *Device) NewPeer(pk NoisePublicKey) (*Peer, error) {
+
+ if device.isClosed.Get() {
+ return nil, errors.New("Device closed")
+ }
+
+ device.mutex.Lock()
+ defer device.mutex.Unlock()
+
// create peer
peer := new(Peer)
peer.mac.Init(pk)
peer.device = device
+ peer.isRunning.Set(false)
+ peer.timer.zeroAllKeys = NewTimer()
peer.timer.keepalivePersistent = NewTimer()
peer.timer.keepalivePassive = NewTimer()
- peer.timer.zeroAllKeys = NewTimer()
peer.timer.handshakeNew = NewTimer()
peer.timer.handshakeDeadline = NewTimer()
peer.timer.handshakeTimeout = NewTimer()
// assign id for debugging
- device.mutex.Lock()
peer.id = device.idCounter
device.idCounter += 1
return nil, errors.New("Adding existing peer")
}
device.peers[pk] = peer
- device.mutex.Unlock()
// precompute DH
peer.endpoint = nil
- // prepare queuing
-
- peer.queue.nonce = make(chan *QueueOutboundElement, QueueOutboundSize)
- peer.queue.outbound = make(chan *QueueOutboundElement, QueueOutboundSize)
- peer.queue.inbound = make(chan *QueueInboundElement, QueueInboundSize)
-
// prepare signaling & routines
- peer.signal.newKeyPair = NewSignal()
- peer.signal.handshakeBegin = NewSignal()
- peer.signal.handshakeCompleted = NewSignal()
- peer.signal.flushNonceQueue = NewSignal()
-
peer.routines.mutex.Lock()
peer.routines.stop = NewSignal()
peer.routines.mutex.Unlock()
+ // start peer
+
+ peer.device.state.mutex.Lock()
+ if peer.device.isUp.Get() {
+ peer.Start()
+ }
+ peer.device.state.mutex.Unlock()
+
return peer, nil
}
return errors.New("No known endpoint for peer")
}
+ if peer.device.net.bind == nil {
+ return errors.New("No bind")
+ }
+
return peer.device.net.bind.Send(buffer, peer.endpoint)
}
peer.routines.mutex.Lock()
defer peer.routines.mutex.Lock()
+ peer.device.log.Debug.Println("Starting:", peer.String())
+
// stop & wait for ungoing routines (if any)
+ peer.isRunning.Set(false)
peer.routines.stop.Broadcast()
peer.routines.starting.Wait()
peer.routines.stopping.Wait()
+ // prepare queues
+
+ peer.signal.newKeyPair = NewSignal()
+ peer.signal.handshakeBegin = NewSignal()
+ peer.signal.handshakeCompleted = NewSignal()
+ peer.signal.flushNonceQueue = NewSignal()
+
+ peer.queue.nonce = make(chan *QueueOutboundElement, QueueOutboundSize)
+ peer.queue.outbound = make(chan *QueueOutboundElement, QueueOutboundSize)
+ peer.queue.inbound = make(chan *QueueInboundElement, QueueInboundSize)
+
// reset signal and start (new) routines
peer.routines.stop = NewSignal()
go peer.RoutineSequentialReceiver()
peer.routines.starting.Wait()
+ peer.isRunning.Set(true)
}
func (peer *Peer) Stop() {
peer.routines.mutex.Lock()
defer peer.routines.mutex.Lock()
+ peer.device.log.Debug.Println("Stopping:", peer.String())
+
// stop & wait for ungoing routines (if any)
peer.routines.stop.Broadcast()
peer.routines.starting.Wait()
peer.routines.stopping.Wait()
+ // close queues
+
+ close(peer.queue.nonce)
+ close(peer.queue.outbound)
+ close(peer.queue.inbound)
+
// reset signal (to handle repeated stopping)
peer.routines.stop = NewSignal()
+ peer.isRunning.Set(false)
}
case ipv6.Version:
size, endpoint, err = bind.ReceiveIPv6(buffer[:])
default:
- return
+ panic("invalid IP version")
}
if err != nil {
// add to decryption queues
- device.addToDecryptionQueue(device.queue.decryption, elem)
- device.addToInboundQueue(peer.queue.inbound, elem)
- buffer = device.GetMessageBuffer()
+ if peer.isRunning.Get() {
+ device.addToDecryptionQueue(device.queue.decryption, elem)
+ device.addToInboundQueue(peer.queue.inbound, elem)
+ buffer = device.GetMessageBuffer()
+ }
continue
return
}
- // lookup peer and consume response
+ // lookup peer from index
entry := device.indices.Lookup(reply.Receiver)
+
if entry.peer == nil {
continue
}
- entry.peer.mac.ConsumeReply(&reply)
+
+ // consume reply
+
+ if peer := entry.peer; peer.isRunning.Get() {
+ peer.mac.ConsumeReply(&reply)
+ }
+
continue
case MessageInitiationType, MessageResponseType:
// insert into nonce/pre-handshake queue
- peer.timer.handshakeDeadline.Reset(RekeyAttemptTime)
- addToOutboundQueue(peer.queue.nonce, elem)
- elem = device.NewOutboundElement()
+ if peer.isRunning.Get() {
+ peer.timer.handshakeDeadline.Reset(RekeyAttemptTime)
+ addToOutboundQueue(peer.queue.nonce, elem)
+ elem = device.NewOutboundElement()
+ }
}
}
// update port and rebind
- device.mutex.Lock()
device.net.mutex.Lock()
-
device.net.port = uint16(port)
- err = unsafeUpdateBind(device)
-
device.net.mutex.Unlock()
- device.mutex.Unlock()
- if err != nil {
+ if err := device.BindUpdate(); err != nil {
logError.Println("Failed to set listen_port:", err)
return &IPCError{Code: ipcErrorPortInUse}
}
device.net.fwmark = uint32(fwmark)
device.net.mutex.Unlock()
+ if err := device.BindUpdate(); err != nil {
+ logError.Println("Failed to update fwmark:", err)
+ return &IPCError{Code: ipcErrorPortInUse}
+ }
+
case "public_key":
// switch to peer configuration
deviceConfig = false
projects / thirdparty / wireguard-go.git / commitdiff
