diff --git a/agent.go b/agent.go
index 5350d1d2..58c204ee 100644
--- a/agent.go
+++ b/agent.go
@@ -42,6 +42,9 @@ type bindingRequest struct {
 type Agent struct {
 	loop *taskloop.Loop
 
+	startedCandidatesMu sync.Mutex
+	startedCandidates   map[*candidateBase]struct{}
+
 	// constructed is set to true after the agent is fully initialized.
 	// Options can check this flag to reject updates that are only valid during construction.
 	constructed bool
@@ -353,6 +356,7 @@ func createAgentBase(config *AgentConfig) (*Agent, error) {
 		lite:                            config.Lite,
 		gatheringState:                  GatheringStateNew,
 		connectionState:                 ConnectionStateNew,
+		startedCandidates:               make(map[*candidateBase]struct{}),
 		localCandidates:                 make(map[NetworkType][]Candidate),
 		remoteCandidates:                make(map[NetworkType][]Candidate),
 		pairsByID:                       make(map[uint64]*CandidatePair),
@@ -1485,7 +1489,7 @@ func (a *Agent) GracefulClose() error {
 
 func (a *Agent) close(graceful bool) error {
 	// the loop is safe to wait on no matter what
-	a.loop.Close()
+	a.loop.CloseWithPreStop(a.abortStartedCandidateIO)
 
 	// but we are in less control of the notifiers, so we will
 	// pass through `graceful`.
@@ -1496,6 +1500,36 @@ func (a *Agent) close(graceful bool) error {
 	return nil
 }
 
+func (a *Agent) registerStartedCandidate(c *candidateBase) {
+	a.startedCandidatesMu.Lock()
+	defer a.startedCandidatesMu.Unlock()
+
+	if a.startedCandidates == nil {
+		a.startedCandidates = make(map[*candidateBase]struct{})
+	}
+	a.startedCandidates[c] = struct{}{}
+}
+
+func (a *Agent) unregisterStartedCandidate(c *candidateBase) {
+	a.startedCandidatesMu.Lock()
+	defer a.startedCandidatesMu.Unlock()
+
+	delete(a.startedCandidates, c)
+}
+
+func (a *Agent) abortStartedCandidateIO() {
+	a.startedCandidatesMu.Lock()
+	candidates := make([]*candidateBase, 0, len(a.startedCandidates))
+	for c := range a.startedCandidates {
+		candidates = append(candidates, c)
+	}
+	a.startedCandidatesMu.Unlock()
+
+	for _, c := range candidates {
+		_ = c.abortIO()
+	}
+}
+
 // Remove all candidates. This closes any listening sockets
 // and removes both the local and remote candidate lists.
 //
diff --git a/agent_test.go b/agent_test.go
index 18f10b03..3413e18f 100644
--- a/agent_test.go
+++ b/agent_test.go
@@ -7,10 +7,13 @@ package ice
 
 import (
 	"context"
+	"io"
 	"net"
 	"net/netip"
+	"os"
 	"strconv"
 	"sync"
+	"sync/atomic"
 	"testing"
 	"time"
 
@@ -51,6 +54,328 @@ func (r *recordingSelector) HandleBindingRequest(*stun.Message, Candidate, Candi
 	r.handledBindingRequest = true
 }
 
+type blockingWritePacketConn struct {
+	writeStarted     chan struct{}
+	writeStartedOnce sync.Once
+	closed           chan struct{}
+	closeOnce        sync.Once
+}
+
+func newBlockingWritePacketConn() *blockingWritePacketConn {
+	return &blockingWritePacketConn{
+		writeStarted: make(chan struct{}),
+		closed:       make(chan struct{}),
+	}
+}
+
+func (b *blockingWritePacketConn) ReadFrom([]byte) (int, net.Addr, error) {
+	<-b.closed
+
+	return 0, nil, io.EOF
+}
+
+func (b *blockingWritePacketConn) WriteTo([]byte, net.Addr) (int, error) {
+	b.writeStartedOnce.Do(func() {
+		close(b.writeStarted)
+	})
+
+	<-b.closed
+
+	return 0, io.ErrClosedPipe
+}
+
+func (b *blockingWritePacketConn) Close() error {
+	b.closeOnce.Do(func() {
+		close(b.closed)
+	})
+
+	return nil
+}
+
+func (b *blockingWritePacketConn) LocalAddr() net.Addr {
+	return &net.UDPAddr{IP: net.IPv4(192, 0, 2, 1), Port: 1}
+}
+
+func (b *blockingWritePacketConn) SetDeadline(time.Time) error {
+	return nil
+}
+
+func (b *blockingWritePacketConn) SetReadDeadline(time.Time) error {
+	return nil
+}
+
+func (b *blockingWritePacketConn) SetWriteDeadline(time.Time) error {
+	return nil
+}
+
+type deadlineBlockingPacketConn struct {
+	*blockingWritePacketConn
+	writeDeadlineSet  chan struct{}
+	writeDeadlineOnce sync.Once
+}
+
+func newDeadlineBlockingPacketConn() *deadlineBlockingPacketConn {
+	return &deadlineBlockingPacketConn{
+		blockingWritePacketConn: newBlockingWritePacketConn(),
+		writeDeadlineSet:        make(chan struct{}),
+	}
+}
+
+func (b *deadlineBlockingPacketConn) WriteTo([]byte, net.Addr) (int, error) {
+	b.writeStartedOnce.Do(func() {
+		close(b.writeStarted)
+	})
+
+	select {
+	case <-b.closed:
+		return 0, io.ErrClosedPipe
+	case <-b.writeDeadlineSet:
+		return 0, os.ErrDeadlineExceeded
+	}
+}
+
+func (b *deadlineBlockingPacketConn) SetDeadline(t time.Time) error {
+	return b.SetWriteDeadline(t)
+}
+
+func (b *deadlineBlockingPacketConn) SetWriteDeadline(t time.Time) error {
+	if !t.IsZero() {
+		b.writeDeadlineOnce.Do(func() {
+			close(b.writeDeadlineSet)
+		})
+	}
+
+	return nil
+}
+
+type blockingMuxedPacketConn struct {
+	*blockingWritePacketConn
+	refs       atomic.Int32
+	closeCount atomic.Int32
+}
+
+func newBlockingMuxedPacketConn() *blockingMuxedPacketConn {
+	return &blockingMuxedPacketConn{
+		blockingWritePacketConn: newBlockingWritePacketConn(),
+	}
+}
+
+func (b *blockingMuxedPacketConn) readFromContext(ctx context.Context, _ []byte) (int, net.Addr, error) {
+	select {
+	case <-ctx.Done():
+		return 0, nil, ctx.Err()
+	case <-b.closed:
+		return 0, nil, io.EOF
+	}
+}
+
+func (b *blockingMuxedPacketConn) Close() error {
+	b.closeCount.Add(1)
+
+	return b.blockingWritePacketConn.Close()
+}
+
+func TestAgentCloseAbortsBlockedCandidateWrite(t *testing.T) {
+	defer test.CheckRoutines(t)()
+	defer test.TimeOut(2 * time.Second).Stop()
+
+	agent, err := NewAgent(&AgentConfig{})
+	require.NoError(t, err)
+
+	conn := newBlockingWritePacketConn()
+	local, err := NewCandidateHost(&CandidateHostConfig{
+		Network:   NetworkTypeUDP4.String(),
+		Address:   "192.0.2.1",
+		Port:      1,
+		Component: ComponentRTP,
+	})
+	require.NoError(t, err)
+	local.start(agent, conn, agent.startedCh)
+
+	remote, err := NewCandidateHost(&CandidateHostConfig{
+		Network:   NetworkTypeUDP4.String(),
+		Address:   "192.0.2.2",
+		Port:      2,
+		Component: ComponentRTP,
+	})
+	require.NoError(t, err)
+
+	msg, err := stun.Build(stun.BindingRequest, stun.TransactionID)
+	require.NoError(t, err)
+
+	runErr := make(chan error, 1)
+	go func() {
+		runErr <- agent.loop.Run(agent.loop, func(context.Context) {
+			agent.sendSTUN(msg, local, remote)
+		})
+	}()
+
+	select {
+	case <-conn.writeStarted:
+	case <-time.After(time.Second):
+		require.Fail(t, "timed out waiting for candidate write to block")
+	}
+
+	closeErr := make(chan error, 1)
+	go func() {
+		closeErr <- agent.Close()
+	}()
+
+	select {
+	case <-conn.closed:
+	case <-time.After(time.Second):
+		require.Fail(t, "timed out waiting for close to abort candidate I/O")
+	}
+
+	select {
+	case err := <-closeErr:
+		require.NoError(t, err)
+	case <-time.After(time.Second):
+		require.Fail(t, "timed out waiting for agent close")
+	}
+
+	require.NoError(t, <-runErr)
+}
+
+func TestAgentCloseAbortsBlockedSharedCandidateWrite(t *testing.T) {
+	defer test.CheckRoutines(t)()
+	defer test.TimeOut(2 * time.Second).Stop()
+
+	agent, err := NewAgent(&AgentConfig{})
+	require.NoError(t, err)
+
+	muxedConn := newBlockingMuxedPacketConn()
+	localA, err := NewCandidateHost(&CandidateHostConfig{
+		Network:   NetworkTypeUDP4.String(),
+		Address:   "192.0.2.1",
+		Port:      1,
+		Component: ComponentRTP,
+	})
+	require.NoError(t, err)
+	localA.start(agent, newSharedPacketConn(muxedConn, &muxedConn.refs), agent.startedCh)
+
+	localB, err := NewCandidateHost(&CandidateHostConfig{
+		Network:   NetworkTypeUDP4.String(),
+		Address:   "192.0.2.1",
+		Port:      2,
+		Component: ComponentRTP,
+	})
+	require.NoError(t, err)
+	localB.start(agent, newSharedPacketConn(muxedConn, &muxedConn.refs), agent.startedCh)
+
+	remote, err := NewCandidateHost(&CandidateHostConfig{
+		Network:   NetworkTypeUDP4.String(),
+		Address:   "192.0.2.2",
+		Port:      3,
+		Component: ComponentRTP,
+	})
+	require.NoError(t, err)
+
+	msg, err := stun.Build(stun.BindingRequest, stun.TransactionID)
+	require.NoError(t, err)
+
+	runErr := make(chan error, 1)
+	go func() {
+		runErr <- agent.loop.Run(agent.loop, func(context.Context) {
+			agent.sendSTUN(msg, localA, remote)
+		})
+	}()
+
+	select {
+	case <-muxedConn.writeStarted:
+	case <-time.After(time.Second):
+		require.Fail(t, "timed out waiting for shared candidate write to block")
+	}
+
+	closeErr := make(chan error, 1)
+	go func() {
+		closeErr <- agent.Close()
+	}()
+
+	select {
+	case <-muxedConn.closed:
+	case <-time.After(time.Second):
+		require.Fail(t, "timed out waiting for close to abort shared candidate I/O")
+	}
+
+	select {
+	case err := <-closeErr:
+		require.NoError(t, err)
+	case <-time.After(time.Second):
+		require.Fail(t, "timed out waiting for agent close")
+	}
+
+	require.NoError(t, <-runErr)
+	require.Equal(t, int32(0), muxedConn.refs.Load())
+	require.Equal(t, int32(1), muxedConn.closeCount.Load())
+}
+
+func TestAgentCloseAbortsBlockedUDPMuxWrite(t *testing.T) {
+	defer test.CheckRoutines(t)()
+	defer test.TimeOut(2 * time.Second).Stop()
+
+	agent, err := NewAgent(&AgentConfig{})
+	require.NoError(t, err)
+
+	udpConn := newDeadlineBlockingPacketConn()
+	udpMux := NewUDPMuxDefault(UDPMuxParams{UDPConn: udpConn})
+	defer func() {
+		_ = udpMux.Close()
+	}()
+
+	muxedConn, err := udpMux.GetConn(agent.localUfrag, udpConn.LocalAddr())
+	require.NoError(t, err)
+
+	local, err := NewCandidateHost(&CandidateHostConfig{
+		Network:   NetworkTypeUDP4.String(),
+		Address:   "192.0.2.1",
+		Port:      1,
+		Component: ComponentRTP,
+	})
+	require.NoError(t, err)
+	local.start(agent, muxedConn, agent.startedCh)
+
+	remote, err := NewCandidateHost(&CandidateHostConfig{
+		Network:   NetworkTypeUDP4.String(),
+		Address:   "192.0.2.2",
+		Port:      2,
+		Component: ComponentRTP,
+	})
+	require.NoError(t, err)
+
+	msg, err := stun.Build(stun.BindingRequest, stun.TransactionID)
+	require.NoError(t, err)
+
+	runErr := make(chan error, 1)
+	go func() {
+		runErr <- agent.loop.Run(agent.loop, func(context.Context) {
+			agent.sendSTUN(msg, local, remote)
+		})
+	}()
+
+	select {
+	case <-udpConn.writeStarted:
+	case <-time.After(time.Second):
+		require.Fail(t, "timed out waiting for UDP mux write to block")
+	}
+
+	closeErr := make(chan error, 1)
+	go func() {
+		closeErr <- agent.Close()
+	}()
+
+	select {
+	case err := <-closeErr:
+		require.NoError(t, err)
+	case <-time.After(time.Second):
+		_ = udpConn.Close()
+		<-closeErr
+		require.Fail(t, "agent close did not abort blocked UDP mux write")
+	}
+
+	require.NoError(t, <-runErr)
+}
+
 func TestHandlePeerReflexive(t *testing.T) { //nolint:cyclop,maintidx
 	defer test.CheckRoutines(t)()
 
diff --git a/candidate_base.go b/candidate_base.go
index 249fe10b..12cfb1d5 100644
--- a/candidate_base.go
+++ b/candidate_base.go
@@ -39,6 +39,8 @@ type candidateBase struct {
 	currAgent *Agent
 	closeCh   chan struct{}
 	closedCh  chan struct{}
+	closeOnce sync.Once
+	closeErr  error
 
 	foundationOverride string
 	priorityOverride   uint32
@@ -50,6 +52,10 @@ type candidateBase struct {
 	extensions            []CandidateExtension
 }
 
+type writeAborter interface {
+	abortWrite() error
+}
+
 // Save a time reference to calculate monotonic time for candidate last sent/received.
 // nolint: gochecknoglobals
 var timeRef = time.Now()
@@ -242,6 +248,7 @@ func (c *candidateBase) start(a *Agent, conn net.PacketConn, initializedCh <-cha
 	c.conn = conn
 	c.closeCh = make(chan struct{})
 	c.closedCh = make(chan struct{})
+	a.registerStartedCandidate(c)
 
 	go c.recvLoop(initializedCh)
 }
@@ -379,34 +386,46 @@ func (c *candidateBase) close() error {
 		return nil
 	}
 
-	// Assert that conn has not already been closed
-	select {
-	case <-c.Done():
-		return nil
-	default:
-	}
+	err := c.abortIO()
 
-	var firstErr error
+	// Wait until the recvLoop is closed
+	<-c.closedCh
 
-	// Unblock recvLoop
-	close(c.closeCh)
-	if err := c.conn.SetDeadline(time.Now()); err != nil {
-		firstErr = err
+	if c.currAgent != nil {
+		c.currAgent.unregisterStartedCandidate(c)
 	}
 
-	// Close the conn
-	if err := c.conn.Close(); err != nil && firstErr == nil {
-		firstErr = err
-	}
+	return err
+}
 
-	if firstErr != nil {
-		return firstErr
+func (c *candidateBase) abortIO() error {
+	// If conn has never been started will be nil.
+	if c.Done() == nil {
+		return nil
 	}
 
-	// Wait until the recvLoop is closed
-	<-c.closedCh
+	c.closeOnce.Do(func() {
+		// Unblock recvLoop.
+		close(c.closeCh)
 
-	return nil
+		if err := c.conn.SetDeadline(time.Now()); err != nil {
+			c.closeErr = err
+		}
+
+		if aborter, ok := c.conn.(writeAborter); ok {
+			if err := aborter.abortWrite(); err != nil && c.closeErr == nil {
+				c.closeErr = err
+			}
+		}
+
+		// Close the conn to abort in-flight socket I/O before the task loop
+		// waits for the current task to finish.
+		if err := c.conn.Close(); err != nil && c.closeErr == nil {
+			c.closeErr = err
+		}
+	})
+
+	return c.closeErr
 }
 
 func (c *candidateBase) writeTo(raw []byte, dst Candidate) (int, error) {
diff --git a/internal/taskloop/taskloop.go b/internal/taskloop/taskloop.go
index c2a3b33e..12563d35 100644
--- a/internal/taskloop/taskloop.go
+++ b/internal/taskloop/taskloop.go
@@ -8,6 +8,7 @@ package taskloop
 import (
 	"context"
 	"errors"
+	"sync"
 	"time"
 
 	atomicx "github.com/pion/ice/v4/internal/atomic"
@@ -28,6 +29,7 @@ type Loop struct {
 	// State for closing
 	done         chan struct{}
 	taskLoopDone chan struct{}
+	closeOnce    sync.Once
 	err          atomicx.Error
 }
 
@@ -65,13 +67,21 @@ func (l *Loop) runLoop(onClose func()) {
 // Close stops the loop after finishing the execution of the current task.
 // Other pending tasks will not be executed.
 func (l *Loop) Close() {
-	if err := l.Err(); err != nil {
-		return
-	}
+	l.CloseWithPreStop(nil)
+}
 
-	l.err.Store(ErrClosed)
+// CloseWithPreStop stops the loop after finishing the execution of the current task.
+// It calls preStop after the loop is marked closed and before waiting for the
+// current task to return.
+func (l *Loop) CloseWithPreStop(preStop func()) {
+	l.closeOnce.Do(func() {
+		l.err.Store(ErrClosed)
 
-	close(l.done)
+		close(l.done)
+		if preStop != nil {
+			preStop()
+		}
+	})
 	<-l.taskLoopDone
 }
 
diff --git a/internal/taskloop/taskloop_test.go b/internal/taskloop/taskloop_test.go
index 06743ce0..1cb8b4ad 100644
--- a/internal/taskloop/taskloop_test.go
+++ b/internal/taskloop/taskloop_test.go
@@ -5,6 +5,7 @@ package taskloop
 
 import (
 	"context"
+	"sync"
 	"sync/atomic"
 	"testing"
 	"time"
@@ -58,3 +59,47 @@ func TestRunReturnsErrClosedWhenLoopClosing(t *testing.T) {
 
 	assert.False(t, secondRan.Load(), "second task should not excute after loop is closed")
 }
+
+func TestCloseWithPreStopConcurrentWaits(t *testing.T) {
+	loop := New(func() {})
+
+	blockStarted := make(chan struct{})
+	releaseBlock := make(chan struct{})
+	go func() {
+		_ = loop.Run(context.Background(), func(context.Context) {
+			close(blockStarted)
+			<-releaseBlock
+		})
+	}()
+	<-blockStarted
+
+	const closers = 8
+	var preStopCalls atomic.Int32
+	closeReturned := make(chan struct{}, closers)
+	var wg sync.WaitGroup
+	wg.Add(closers)
+	for range closers {
+		go func() {
+			defer wg.Done()
+			loop.CloseWithPreStop(func() {
+				preStopCalls.Add(1)
+			})
+			closeReturned <- struct{}{}
+		}()
+	}
+
+	assert.Eventually(t, func() bool {
+		return preStopCalls.Load() == 1
+	}, time.Second, time.Millisecond)
+
+	select {
+	case <-closeReturned:
+		assert.Fail(t, "CloseWithPreStop returned before the active task finished")
+	case <-time.After(10 * time.Millisecond):
+	}
+
+	close(releaseBlock)
+	wg.Wait()
+
+	assert.Equal(t, int32(1), preStopCalls.Load())
+}
diff --git a/shared_packet_conn.go b/shared_packet_conn.go
index ec6f5157..43cc576c 100644
--- a/shared_packet_conn.go
+++ b/shared_packet_conn.go
@@ -114,6 +114,15 @@ func (s *sharedPacketConn) SetDeadline(t time.Time) error {
 	return s.SetWriteDeadline(t)
 }
 
+func (s *sharedPacketConn) abortWrite() error {
+	aborter, ok := s.underlying.(writeAborter)
+	if !ok {
+		return nil
+	}
+
+	return aborter.abortWrite()
+}
+
 func (s *sharedPacketConn) Close() error {
 	var err error
 	fired := false
diff --git a/udp_mux.go b/udp_mux.go
index aa628aee..6b49e551 100644
--- a/udp_mux.go
+++ b/udp_mux.go
@@ -9,8 +9,11 @@ import (
 	"net"
 	"net/netip"
 	"os"
+	"runtime"
 	"strings"
 	"sync"
+	"sync/atomic"
+	"time"
 
 	"github.com/pion/logging"
 	"github.com/pion/stun/v3"
@@ -46,8 +49,17 @@ type UDPMuxDefault struct {
 
 	// whether the UDP connection listens on an unspecified address
 	isUnspecified bool
+
+	// writeState packs write in-flight count.
+	writeState atomic.Uint64
 }
 
+const (
+	udpMuxWriteBlockedBit  = uint64(1) << 63
+	udpMuxWriteDeadlineBit = uint64(1) << 62
+	udpMuxWriteCountMask   = udpMuxWriteDeadlineBit - 1
+)
+
 // UDPMuxParams are parameters for UDPMux.
 type UDPMuxParams struct {
 	Logger        logging.LeveledLogger
@@ -98,7 +110,6 @@ func NewUDPMuxDefault(params UDPMuxParams) *UDPMuxDefault {
 		},
 		isUnspecified: isUnspecified,
 	}
-
 	go mux.connWorker()
 
 	return mux
@@ -258,9 +269,122 @@ func (m *UDPMuxDefault) Close() error {
 }
 
 func (m *UDPMuxDefault) writeTo(buf []byte, rAddr net.Addr) (n int, err error) {
+	m.startWrite()
+
+	defer func() {
+		err = m.finishWrite(err)
+	}()
+
 	return m.params.UDPConn.WriteTo(buf, rAddr)
 }
 
+func (m *UDPMuxDefault) abortWrite() error {
+	for {
+		state := m.writeState.Load()
+		if state&udpMuxWriteBlockedBit != 0 || state&udpMuxWriteCountMask == 0 {
+			return nil
+		}
+
+		if !m.writeState.CompareAndSwap(state, state|udpMuxWriteBlockedBit) {
+			continue
+		}
+
+		if err := m.params.UDPConn.SetWriteDeadline(time.Now()); err != nil {
+			m.clearWriteAbortState()
+
+			return err
+		}
+
+		m.setWriteDeadlineArmed()
+
+		return nil
+	}
+}
+
+func (m *UDPMuxDefault) startWrite() {
+	for {
+		state := m.writeState.Load()
+		if state&udpMuxWriteBlockedBit != 0 {
+			runtime.Gosched()
+
+			continue
+		}
+
+		if m.writeState.CompareAndSwap(state, state+1) {
+			return
+		}
+	}
+}
+
+func (m *UDPMuxDefault) finishWrite(writeErr error) error {
+	for {
+		state := m.writeState.Load()
+		count := state & udpMuxWriteCountMask
+		if count == 0 {
+			return writeErr
+		}
+
+		if state&udpMuxWriteBlockedBit != 0 && count == 1 {
+			if !m.writeState.CompareAndSwap(state, state-1) {
+				continue
+			}
+
+			return m.clearWriteDeadlineAfterAbort(writeErr)
+		}
+
+		if m.writeState.CompareAndSwap(state, state-1) {
+			return writeErr
+		}
+	}
+}
+
+func (m *UDPMuxDefault) setWriteDeadlineArmed() {
+	for {
+		state := m.writeState.Load()
+		if state&udpMuxWriteBlockedBit == 0 || state&udpMuxWriteDeadlineBit != 0 {
+			return
+		}
+		if m.writeState.CompareAndSwap(state, state|udpMuxWriteDeadlineBit) {
+			return
+		}
+	}
+}
+
+func (m *UDPMuxDefault) clearWriteDeadlineAfterAbort(writeErr error) error {
+	for {
+		state := m.writeState.Load()
+		if state&udpMuxWriteBlockedBit == 0 {
+			return writeErr
+		}
+		if state&udpMuxWriteDeadlineBit == 0 {
+			runtime.Gosched()
+
+			continue
+		}
+
+		clearErr := m.params.UDPConn.SetWriteDeadline(time.Time{})
+		m.writeState.Store(0)
+		if writeErr == nil {
+			return clearErr
+		}
+
+		return writeErr
+	}
+}
+
+func (m *UDPMuxDefault) clearWriteAbortState() {
+	for {
+		state := m.writeState.Load()
+		newState := state &^ (udpMuxWriteBlockedBit | udpMuxWriteDeadlineBit)
+		if state == newState {
+			return
+		}
+		if m.writeState.CompareAndSwap(state, newState) {
+			return
+		}
+	}
+}
+
 func (m *UDPMuxDefault) registerConnForAddress(conn *udpMuxedConn, addr ipPort) {
 	if m.IsClosed() {
 		return
diff --git a/udp_muxed_conn.go b/udp_muxed_conn.go
index c70416c8..31937aea 100644
--- a/udp_muxed_conn.go
+++ b/udp_muxed_conn.go
@@ -146,6 +146,10 @@ func (c *udpMuxedConn) SetWriteDeadline(time.Time) error {
 	return nil
 }
 
+func (c *udpMuxedConn) abortWrite() error {
+	return c.params.Mux.abortWrite()
+}
+
 func (c *udpMuxedConn) CloseChannel() <-chan struct{} {
 	return c.closedChan
 }