Block for link address resolution

Previously, if address resolution for UDP or Ping sockets required sending
packets using Write in Transport layer, Resolve would return ErrWouldBlock
and Write would return ErrNoLinkAddress. Meanwhile startAddressResolution
would run in background. Further calls to Write using same address would also
return ErrNoLinkAddress until resolution has been completed successfully.

Since Write is not allowed to block and System Calls need to be
interruptible in System Call layer, the caller to Write is responsible for
blocking upon return of ErrWouldBlock.

Now, when startAddressResolution is called a notification channel for
the completion of the address resolution is returned.
The channel will traverse up to the calling function of Write as well as
ErrNoLinkAddress. Once address resolution is complete (success or not) the
channel is closed. The caller would call Write again to send packets and
check if address resolution was compeleted successfully or not.

Fixes google#5

Change-Id: Idafaf31982bee1915ca084da39ae7bd468cebd93
PiperOrigin-RevId: 214962200

Author: sepehrtheraiss

pkg/dhcp/client.go

@@ -195,10 +195,23 @@ func (c *Client) Request(ctx context.Context, requestedAddr tcpip.Address) (cfg
 	wopts := tcpip.WriteOptions{
 		To: serverAddr,
 	}
-	if _, err := ep.Write(tcpip.SlicePayload(h), wopts); err != nil {
+	var resCh <-chan struct{}
+	if _, resCh, err = ep.Write(tcpip.SlicePayload(h), wopts); err != nil && resCh == nil {
 		return Config{}, fmt.Errorf("dhcp discovery write: %v", err)
 	}
 
+	if resCh != nil {
+		select {
+		case <-resCh:
+		case <-ctx.Done():
+			return Config{}, fmt.Errorf("dhcp client address resolution: %v", tcpip.ErrAborted)
+		}
+
+		if _, _, err := ep.Write(tcpip.SlicePayload(h), wopts); err != nil {
+			return Config{}, fmt.Errorf("dhcp discovery write: %v", err)
+		}
+	}
+
 	we, ch := waiter.NewChannelEntry(nil)
 	wq.EventRegister(&we, waiter.EventIn)
 	defer wq.EventUnregister(&we)
@@ -289,7 +302,7 @@ func (c *Client) Request(ctx context.Context, requestedAddr tcpip.Address) (cfg
 		reqOpts = append(reqOpts, option{optClientID, clientID})
 	}
 	h.setOptions(reqOpts)
-	if _, err := ep.Write(tcpip.SlicePayload(h), wopts); err != nil {
+	if _, _, err := ep.Write(tcpip.SlicePayload(h), wopts); err != nil {
 		return Config{}, fmt.Errorf("dhcp discovery write: %v", err)
 	}
 

pkg/dhcp/server.go

@@ -95,9 +95,22 @@ func (c *epConn) Read() (buffer.View, tcpip.FullAddress, error) {
 }
 
 func (c *epConn) Write(b []byte, addr *tcpip.FullAddress) error {
-	if _, err := c.ep.Write(tcpip.SlicePayload(b), tcpip.WriteOptions{To: addr}); err != nil {
+	_, resCh, err := c.ep.Write(tcpip.SlicePayload(b), tcpip.WriteOptions{To: addr})
+	if err != nil && resCh == nil {
 		return fmt.Errorf("write: %v", err)
 	}
+
+	if resCh != nil {
+		select {
+		case <-resCh:
+		case <-c.ctx.Done():
+			return fmt.Errorf("dhcp server address resolution: %v", tcpip.ErrAborted)
+		}
+
+		if _, _, err := c.ep.Write(tcpip.SlicePayload(b), tcpip.WriteOptions{To: addr}); err != nil {
+			return fmt.Errorf("write: %v", err)
+		}
+	}
 	return nil
 }
 

pkg/sentry/socket/epsocket/epsocket.go

@@ -276,10 +276,21 @@ func (i *ioSequencePayload) Size() int {
 // Write implements fs.FileOperations.Write.
 func (s *SocketOperations) Write(ctx context.Context, _ *fs.File, src usermem.IOSequence, _ int64) (int64, error) {
 	f := &ioSequencePayload{ctx: ctx, src: src}
-	n, err := s.Endpoint.Write(f, tcpip.WriteOptions{})
+	n, resCh, err := s.Endpoint.Write(f, tcpip.WriteOptions{})
 	if err == tcpip.ErrWouldBlock {
 		return int64(n), syserror.ErrWouldBlock
 	}
+
+	if resCh != nil {
+		t := ctx.(*kernel.Task)
+		if err := t.Block(resCh); err != nil {
+			return int64(n), syserr.FromError(err).ToError()
+		}
+
+		n, _, err = s.Endpoint.Write(f, tcpip.WriteOptions{})
+		return int64(n), syserr.TranslateNetstackError(err).ToError()
+	}
+
 	return int64(n), syserr.TranslateNetstackError(err).ToError()
 }
 
@@ -1016,7 +1027,13 @@ func (s *SocketOperations) SendMsg(t *kernel.Task, src usermem.IOSequence, to []
 		EndOfRecord: flags&linux.MSG_EOR != 0,
 	}
 
-	n, err := s.Endpoint.Write(tcpip.SlicePayload(v), opts)
+	n, resCh, err := s.Endpoint.Write(tcpip.SlicePayload(v), opts)
+	if resCh != nil {
+		if err := t.Block(resCh); err != nil {
+			return int(n), syserr.FromError(err)
+		}
+		n, _, err = s.Endpoint.Write(tcpip.SlicePayload(v), opts)
+	}
 	if err != tcpip.ErrWouldBlock || flags&linux.MSG_DONTWAIT != 0 {
 		return int(n), syserr.TranslateNetstackError(err)
 	}
@@ -1030,7 +1047,7 @@ func (s *SocketOperations) SendMsg(t *kernel.Task, src usermem.IOSequence, to []
 	v.TrimFront(int(n))
 	total := n
 	for {
-		n, err = s.Endpoint.Write(tcpip.SlicePayload(v), opts)
+		n, _, err = s.Endpoint.Write(tcpip.SlicePayload(v), opts)
 		v.TrimFront(int(n))
 		total += n
 		if err != tcpip.ErrWouldBlock {

pkg/tcpip/adapters/gonet/gonet.go

@@ -393,9 +393,22 @@ func (c *Conn) Write(b []byte) (int, error) {
 		}
 
 		var n uintptr
-		n, err = c.ep.Write(tcpip.SlicePayload(v), tcpip.WriteOptions{})
+		var resCh <-chan struct{}
+		n, resCh, err = c.ep.Write(tcpip.SlicePayload(v), tcpip.WriteOptions{})
 		nbytes += int(n)
 		v.TrimFront(int(n))
+
+		if resCh != nil {
+			select {
+			case <-deadline:
+				return nbytes, c.newOpError("write", &timeoutError{})
+			case <-resCh:
+			}
+
+			n, _, err = c.ep.Write(tcpip.SlicePayload(v), tcpip.WriteOptions{})
+			nbytes += int(n)
+			v.TrimFront(int(n))
+		}
 	}
 
 	if err == nil {
@@ -571,23 +584,33 @@ func (c *PacketConn) WriteTo(b []byte, addr net.Addr) (int, error) {
 	copy(v, b)
 
 	wopts := tcpip.WriteOptions{To: &fullAddr}
-	n, err := c.ep.Write(tcpip.SlicePayload(v), wopts)
+	n, resCh, err := c.ep.Write(tcpip.SlicePayload(v), wopts)
+	if resCh != nil {
+		select {
+		case <-deadline:
+			return int(n), c.newRemoteOpError("write", addr, &timeoutError{})
+		case <-resCh:
+		}
+
+		n, _, err = c.ep.Write(tcpip.SlicePayload(v), wopts)
+	}
 
 	if err == tcpip.ErrWouldBlock {
 		// Create wait queue entry that notifies a channel.
 		waitEntry, notifyCh := waiter.NewChannelEntry(nil)
 		c.wq.EventRegister(&waitEntry, waiter.EventOut)
 		defer c.wq.EventUnregister(&waitEntry)
 		for {
-			n, err = c.ep.Write(tcpip.SlicePayload(v), wopts)
-			if err != tcpip.ErrWouldBlock {
-				break
-			}
 			select {
 			case <-deadline:
 				return int(n), c.newRemoteOpError("write", addr, &timeoutError{})
 			case <-notifyCh:
 			}
+
+			n, _, err = c.ep.Write(tcpip.SlicePayload(v), wopts)
+			if err != tcpip.ErrWouldBlock {
+				break
+			}
 		}
 	}
 

pkg/tcpip/network/ipv6/icmp_test.go

@@ -190,7 +190,7 @@ func TestLinkResolution(t *testing.T) {
 		if ctx.Err() != nil {
 			break
 		}
-		if _, err := ep.Write(payload, tcpip.WriteOptions{To: &tcpip.FullAddress{NIC: 1, Addr: lladdr1}}); err == tcpip.ErrNoLinkAddress {
+		if _, _, err := ep.Write(payload, tcpip.WriteOptions{To: &tcpip.FullAddress{NIC: 1, Addr: lladdr1}}); err == tcpip.ErrNoLinkAddress {
 			// There's something asynchronous going on; yield to let it do its thing.
 			runtime.Gosched()
 		} else if err == nil {

pkg/tcpip/sample/tun_tcp_connect/main.go

@@ -80,7 +80,7 @@ func writer(ch chan struct{}, ep tcpip.Endpoint) {
 
 		v.CapLength(n)
 		for len(v) > 0 {
-			n, err := ep.Write(tcpip.SlicePayload(v), tcpip.WriteOptions{})
+			n, _, err := ep.Write(tcpip.SlicePayload(v), tcpip.WriteOptions{})
 			if err != nil {
 				fmt.Println("Write failed:", err)
 				return

pkg/tcpip/stack/linkaddrcache.go

@@ -88,12 +88,14 @@ type linkAddrEntry struct {
 	linkAddr   tcpip.LinkAddress
 	expiration time.Time
 	s          entryState
+	resDone    bool
 
 	// wakers is a set of waiters for address resolution result. Anytime
 	// state transitions out of 'incomplete' these waiters are notified.
 	wakers map[*sleep.Waker]struct{}
 
 	cancel chan struct{}
+	resCh  chan struct{}
 }
 
 func (e *linkAddrEntry) state() entryState {
@@ -182,15 +184,20 @@ func (c *linkAddrCache) makeAndAddEntry(k tcpip.FullAddress, v tcpip.LinkAddress
 	// someone waiting for address resolution on it.
 	entry.changeState(expired)
 	if entry.cancel != nil {
-		entry.cancel <- struct{}{}
+		if !entry.resDone {
+			close(entry.resCh)
+		}
+		close(entry.cancel)
 	}
 
 	*entry = linkAddrEntry{
 		addr:       k,
 		linkAddr:   v,
 		expiration: time.Now().Add(c.ageLimit),
+		resDone:    false,
 		wakers:     make(map[*sleep.Waker]struct{}),
 		cancel:     make(chan struct{}, 1),
+		resCh:      make(chan struct{}, 1),
 	}
 
 	c.cache[k] = entry
@@ -202,10 +209,10 @@ func (c *linkAddrCache) makeAndAddEntry(k tcpip.FullAddress, v tcpip.LinkAddress
 }
 
 // get reports any known link address for k.
-func (c *linkAddrCache) get(k tcpip.FullAddress, linkRes LinkAddressResolver, localAddr tcpip.Address, linkEP LinkEndpoint, waker *sleep.Waker) (tcpip.LinkAddress, *tcpip.Error) {
+func (c *linkAddrCache) get(k tcpip.FullAddress, linkRes LinkAddressResolver, localAddr tcpip.Address, linkEP LinkEndpoint, waker *sleep.Waker) (tcpip.LinkAddress, <-chan struct{}, *tcpip.Error) {
 	if linkRes != nil {
 		if addr, ok := linkRes.ResolveStaticAddress(k.Addr); ok {
-			return addr, nil
+			return addr, nil, nil
 		}
 	}
 
@@ -214,10 +221,11 @@ func (c *linkAddrCache) get(k tcpip.FullAddress, linkRes LinkAddressResolver, lo
 	if entry == nil || entry.state() == expired {
 		c.mu.Unlock()
 		if linkRes == nil {
-			return "", tcpip.ErrNoLinkAddress
+			return "", nil, tcpip.ErrNoLinkAddress
 		}
-		c.startAddressResolution(k, linkRes, localAddr, linkEP, waker)
-		return "", tcpip.ErrWouldBlock
+
+		ch := c.startAddressResolution(k, linkRes, localAddr, linkEP, waker)
+		return "", ch, tcpip.ErrWouldBlock
 	}
 	defer c.mu.Unlock()
 
@@ -227,13 +235,13 @@ func (c *linkAddrCache) get(k tcpip.FullAddress, linkRes LinkAddressResolver, lo
 		// in that case it's safe to consider it ready.
 		fallthrough
 	case ready:
-		return entry.linkAddr, nil
+		return entry.linkAddr, nil, nil
 	case failed:
-		return "", tcpip.ErrNoLinkAddress
+		return "", nil, tcpip.ErrNoLinkAddress
 	case incomplete:
 		// Address resolution is still in progress.
 		entry.addWaker(waker)
-		return "", tcpip.ErrWouldBlock
+		return "", entry.resCh, tcpip.ErrWouldBlock
 	default:
 		panic(fmt.Sprintf("invalid cache entry state: %d", s))
 	}
@@ -249,13 +257,13 @@ func (c *linkAddrCache) removeWaker(k tcpip.FullAddress, waker *sleep.Waker) {
 	}
 }
 
-func (c *linkAddrCache) startAddressResolution(k tcpip.FullAddress, linkRes LinkAddressResolver, localAddr tcpip.Address, linkEP LinkEndpoint, waker *sleep.Waker) {
+func (c *linkAddrCache) startAddressResolution(k tcpip.FullAddress, linkRes LinkAddressResolver, localAddr tcpip.Address, linkEP LinkEndpoint, waker *sleep.Waker) <-chan struct{} {
 	c.mu.Lock()
 	defer c.mu.Unlock()
 
 	// Look up again with lock held to ensure entry wasn't added by someone else.
 	if e := c.cache[k]; e != nil && e.state() != expired {
-		return
+		return nil
 	}
 
 	// Add 'incomplete' entry in the cache to mark that resolution is in progress.
@@ -274,13 +282,23 @@ func (c *linkAddrCache) startAddressResolution(k tcpip.FullAddress, linkRes Link
 			select {
 			case <-time.After(c.resolutionTimeout):
 				if stop := c.checkLinkRequest(k, i); stop {
+					// If entry is evicted then resCh is already closed.
+					c.mu.Lock()
+					if e, ok := c.cache[k]; ok {
+						if !e.resDone {
+							e.resDone = true
+							close(e.resCh)
+						}
+					}
+					c.mu.Unlock()
 					return
 				}
 			case <-cancel:
 				return
 			}
 		}
 	}()
+	return e.resCh
 }
 
 // checkLinkRequest checks whether previous attempt to resolve address has succeeded