detached 0f188ef

refactor(dgram): split transport lifecycle and send paths

commit
0f188ef23d5b21c7a314b5303c91f2f8184cb29d
author
brice · · 2026-07-14 20:45 UTC
parent
5b6c186
signature
GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email
4 files · +445 -425
Unified
UnifiedSplit
M internal/adapters/dgram/control.go +38 -0
@@ -214,3 +214,41 @@ func (t *Transport) sendControl(kind byte, id uint64) error {
214214 }
215215 return nil
216216 }
217+
218+func (t *Transport) sendAck(seq uint64) {
219+ var b [9]byte
220+ b[0] = recAck
221+ binary.BigEndian.PutUint64(b[1:], seq)
222+ _ = t.sendPayload(b[:])
223+}
224+func (t *Transport) sendPayload(p []byte) error {
225+ frags, err := pdgram.FragmentPayload(t.nextCounter(), p, t.mtu-pdgram.HeaderSize-t.codec.Overhead())
226+ if err != nil {
227+ return err
228+ }
229+ for _, f := range frags {
230+ raw, err := pdgram.MarshalFragment(f)
231+ if err != nil {
232+ return err
233+ }
234+ pkt := t.codec.Seal(t.sendDir, t.nextCounter(), raw, nil)
235+ t.mu.Lock()
236+ peer := t.peer
237+ closed := t.closed
238+ closeErr := t.closeErr
239+ t.mu.Unlock()
240+ if closed {
241+ if closeErr != nil {
242+ return closeErr
243+ }
244+ return errors.New("dgram: closed")
245+ }
246+ if peer == nil {
247+ return errors.New("dgram: no peer")
248+ }
249+ if err := t.writePacket(pkt, peer); err != nil {
250+ return err
251+ }
252+ }
253+ return nil
254+}
@@ -214,3 +214,41 @@ func (t *Transport) sendControl(kind byte, id uint64) error {
214 }214 }
215 return nil215 return nil
216 }216 }
217+
218+func (t *Transport) sendAck(seq uint64) {
219+ var b [9]byte
220+ b[0] = recAck
221+ binary.BigEndian.PutUint64(b[1:], seq)
222+ _ = t.sendPayload(b[:])
223+}
224+func (t *Transport) sendPayload(p []byte) error {
225+ frags, err := pdgram.FragmentPayload(t.nextCounter(), p, t.mtu-pdgram.HeaderSize-t.codec.Overhead())
226+ if err != nil {
227+ return err
228+ }
229+ for _, f := range frags {
230+ raw, err := pdgram.MarshalFragment(f)
231+ if err != nil {
232+ return err
233+ }
234+ pkt := t.codec.Seal(t.sendDir, t.nextCounter(), raw, nil)
235+ t.mu.Lock()
236+ peer := t.peer
237+ closed := t.closed
238+ closeErr := t.closeErr
239+ t.mu.Unlock()
240+ if closed {
241+ if closeErr != nil {
242+ return closeErr
243+ }
244+ return errors.New("dgram: closed")
245+ }
246+ if peer == nil {
247+ return errors.New("dgram: no peer")
248+ }
249+ if err := t.writePacket(pkt, peer); err != nil {
250+ return err
251+ }
252+ }
253+ return nil
254+}
M internal/adapters/dgram/transport.go +0 -425
@@ -324,319 +324,6 @@ func (s *writeDeadlineState) expire(now time.Time) {
324324 s.mu.Unlock()
325325 }
326326
327-func NewTransport(pc net.PacketConn, peer net.Addr, key []byte, sendDir, recvDir uint32, runtimeOpts ...RuntimeOption) (*Transport, error) {
328- return NewTransportWithOptions(pc, peer, key, sendDir, recvDir, Options{}, runtimeOpts...)
329-}
330-
331-func NewTransportWithOptions(pc net.PacketConn, peer net.Addr, key []byte, sendDir, recvDir uint32, opts Options, runtimeOpts ...RuntimeOption) (*Transport, error) {
332- c, err := pdgram.NewCodec(key)
333- if err != nil {
334- return nil, err
335- }
336- opts = normalizeOptions(opts)
337- t := &Transport{
338- pc: pc,
339- codec: c,
340- sendDir: sendDir,
341- recvDir: recvDir,
342- mtu: opts.MTU,
343- peer: peer,
344- pending: make(map[uint64]*pending),
345- replay: pdgram.NewReplayWindow(),
346- reasm: pdgram.NewReassembler(),
347- in: make(chan ports.Frame, 32),
348- done: make(chan struct{}),
349- health: newHealthTracker(opts.Clock.Now()),
350- lastAuthenticatedPacket: opts.Clock.Now(),
351- lastCompleteRecord: opts.Clock.Now(),
352- lastACKProgress: opts.Clock.Now(),
353- observe: opts.Observe,
354- heartbeat: opts.Heartbeat,
355- resendAfter: opts.ResendAfter,
356- maxResendAfter: opts.MaxResendAfter,
357- maxResendPerTick: opts.MaxResendPerTick,
358- rto: opts.ResendAfter,
359- writeTimeout: opts.WriteTimeout,
360- degradedAfter: opts.DegradedAfter,
361- probeAfter: opts.ProbeAfter,
362- offlineAfter: opts.OfflineAfter,
363- deadAfter: opts.DeadAfter,
364- maxPending: opts.MaxPending,
365- maxPendingWait: opts.MaxPendingWait,
366- maxRecvBuffer: opts.MaxRecvBuffer,
367- clock: opts.Clock,
368- congestion: newCongestionController(opts.MTU),
369- linkState: ports.LinkStateConnected,
370- linkEvents: make(chan ports.LinkEvent, linkEventBufferSize),
371- probeWait: make(map[uint64]chan struct{}),
372- rebind: opts.RebindPacketConn,
373- nextRecvSeq: 1,
374- recvBuf: make(map[uint64]ports.Frame),
375- }
376- for _, opt := range runtimeOpts {
377- if opt != nil {
378- opt(t)
379- }
380- }
381- t.sendWake = make(chan struct{})
382- t.control = make(chan controlRecord, controlQueueSize)
383- t.ackWake = make(chan struct{}, 1)
384- t.ackSend = make(chan uint64, 1)
385- t.outputWake = make(chan struct{})
386- t.dataSend = make(chan dataSendJob, opts.MaxPending)
387- t.writeDeadlines = newWriteDeadlineState(pc)
388- // One queued batch bounds retransmit work while its sole sender is paced.
389- t.retransmitWork = make(chan []retransmitRecord, 1)
390- if t.observe != nil {
391- t.diagnosticCh = make(chan Diagnostic, diagnosticBufferSize)
392- go t.diagnosticLoop()
393- }
394- t.deliverCond = sync.NewCond(&t.deliverMu)
395- go t.readLoop(pc)
396- go t.retransmitLoop()
397- go t.heartbeatLoop()
398- go t.healthLoop()
399- go t.dataSendLoop()
400- go t.deliveryLoop()
401- go t.controlLoop()
402- go t.ackSendLoop()
403- go t.outputPaceLoop()
404- t.emitDiagnostic()
405- return t, nil
406-}
407-
408-func (t *Transport) Send(f ports.Frame) error {
409- end := t.beginRuntimeOperation(ports.RuntimeAdapterSendStart, uint64(len(f.Payload)))
410- err := t.send(f, false)
411- end(err == nil)
412- return err
413-}
414-
415-func (t *Transport) SendAsync(f ports.Frame) error {
416- end := t.beginRuntimeOperation(ports.RuntimeAdapterSendStart, uint64(len(f.Payload)))
417- err := t.send(f, true)
418- end(err == nil)
419- return err
420-}
421-
422-// SendSynchronous owns queue reservation, fragment pacing, per-write deadlines,
423-// and close cancellation. Callers must not layer a shorter preflight timer over it.
424-func (t *Transport) SendSynchronous(f ports.Frame) error { return t.Send(f) }
425-
426-func (t *Transport) send(f ports.Frame, async bool) error {
427- reliable := true
428- if err := t.lockOutboundSlot(reliable); err != nil {
429- return err
430- }
431- unlockOutbound := true
432- defer func() {
433- if unlockOutbound {
434- t.outboundMu.Unlock()
435- }
436- }()
437- // lockOutboundSlot returns with both outboundMu and mu held.
438-
439- t.seq++
440- seq := t.seq
441- if reliable {
442- now := t.clock.Now()
443- if len(t.pending) == 0 {
444- t.health.pendingStarted(now)
445- }
446- t.pending[seq] = &pending{frame: f, enqueued: now}
447- }
448- var done chan error
449- if !async {
450- done = make(chan error, 1)
451- }
452- job := dataSendJob{seq: seq, reliable: reliable, frame: f, done: done}
453- if shouldPaceOutput(f) || len(t.outputQueue) > 0 {
454- wasEmpty := len(t.outputQueue) == 0
455- t.outputQueue = append(t.outputQueue, queuedSend(job))
456- if wasEmpty {
457- t.notifyOutputPacerLocked()
458- }
459- t.mu.Unlock()
460- t.outboundMu.Unlock()
461- unlockOutbound = false
462- if async {
463- return nil
464- }
465- return t.waitQueuedSend(done)
466- }
467- t.mu.Unlock()
468- if err := t.queueDataJob(job); err != nil {
469- t.removePending(seq, reliable)
470- return err
471- }
472- t.outboundMu.Unlock()
473- unlockOutbound = false
474- if async {
475- return nil
476- }
477- return t.waitQueuedSend(done)
478-}
479-
480-func (t *Transport) queueDataJob(job dataSendJob) error {
481- select {
482- case t.dataSend <- job:
483- return nil
484- case <-t.done:
485- return t.closedError()
486- }
487-}
488-
489-func (t *Transport) waitQueuedSend(done <-chan error) error {
490- select {
491- case err := <-done:
492- return err
493- case <-t.done:
494- t.mu.Lock()
495- err := t.closeErr
496- t.mu.Unlock()
497- if err != nil {
498- return err
499- }
500- return errors.New("dgram: closed")
501- }
502-}
503-
504-func (t *Transport) closedError() error {
505- t.mu.Lock()
506- defer t.mu.Unlock()
507- if t.closeErr != nil {
508- return t.closeErr
509- }
510- return errors.New("dgram: closed")
511-}
512-
513-func (t *Transport) lockOutboundSlot(reliable bool) error {
514- var pendingWaitStarted time.Time
515- for {
516- t.outboundMu.Lock()
517- t.mu.Lock()
518- if t.closed {
519- err := t.closeErr
520- t.mu.Unlock()
521- t.outboundMu.Unlock()
522- if err != nil {
523- return err
524- }
525- return errors.New("dgram: closed")
526- }
527- if !reliable || len(t.pending) < t.maxPending {
528- return nil
529- }
530- if t.linkState != ports.LinkStateConnected {
531- t.mu.Unlock()
532- t.outboundMu.Unlock()
533- return ErrPendingFull
534- }
535- if pendingWaitStarted.IsZero() {
536- pendingWaitStarted = t.clock.Now()
537- }
538- remaining := t.maxPendingWait - t.clock.Now().Sub(pendingWaitStarted)
539- if remaining <= 0 {
540- t.mu.Unlock()
541- t.outboundMu.Unlock()
542- return ErrPendingFull
543- }
544- wake := t.sendWake
545- timer := t.clock.NewTimer(remaining)
546- t.mu.Unlock()
547- t.outboundMu.Unlock()
548- select {
549- case <-timer.C():
550- case <-wake:
551- case <-t.done:
552- }
553- timer.Stop()
554- }
555-}
556-
557-func (t *Transport) removePending(seq uint64, reliable bool) {
558- if !reliable {
559- return
560- }
561- t.mu.Lock()
562- removed := t.removePendingLocked(seq)
563- if removed {
564- t.notifySendWaitersLocked()
565- }
566- t.mu.Unlock()
567-}
568-
569-func (t *Transport) removeQueuedPending(queued []queuedSend, err error) {
570- t.mu.Lock()
571- removed := false
572- for _, q := range queued {
573- if q.reliable {
574- removed = t.removePendingLocked(q.seq) || removed
575- }
576- if q.done != nil {
577- q.done <- err
578- close(q.done)
579- }
580- }
581- if removed {
582- t.notifySendWaitersLocked()
583- }
584- t.mu.Unlock()
585-}
586-
587-func (t *Transport) removePendingLocked(seq uint64) bool {
588- if _, ok := t.pending[seq]; !ok {
589- return false
590- }
591- delete(t.pending, seq)
592- if len(t.pending) == 0 {
593- t.health.pendingCleared()
594- }
595- return true
596-}
597-
598-func (t *Transport) Recv() (ports.Frame, error) {
599- end := t.beginRuntimeOperation(ports.RuntimeAdapterReceiveStart, 0)
600- select {
601- case f := <-t.in:
602- end(true)
603- return f, nil
604- case <-t.done:
605- t.mu.Lock()
606- err := t.closeErr
607- t.mu.Unlock()
608- if err != nil {
609- end(false)
610- return ports.Frame{}, err
611- }
612- err = errors.New("dgram: closed")
613- end(false)
614- return ports.Frame{}, err
615- }
616-}
617-func (t *Transport) Close() error {
618- done := t.beginShutdown()
619- t.closeWithError(errors.New("dgram: closed"))
620- t.mu.Lock()
621- pc := t.pc
622- t.mu.Unlock()
623- err := pc.Close()
624- if done != nil {
625- <-done
626- }
627- return err
628-}
629-
630-func (t *Transport) Peer() net.Addr { t.mu.Lock(); defer t.mu.Unlock(); return t.peer }
631-
632-func (t *Transport) LinkState() ports.LinkState {
633- t.mu.Lock()
634- defer t.mu.Unlock()
635- return t.linkState
636-}
637-
638-func (t *Transport) LinkEvents() <-chan ports.LinkEvent { return t.linkEvents }
639-
640327 // Probe sends an authenticated datagram and waits for the peer to authenticate a
641328 // response or for ctx to expire. It is intended for UDP bootstrap reachability checks.
642329 func (t *Transport) Probe(ctx context.Context) error {
@@ -694,44 +381,6 @@ func (t *Transport) unregisterProbe(id uint64) {
694381 t.mu.Unlock()
695382 }
696383
697-func (t *Transport) sendAck(seq uint64) {
698- var b [9]byte
699- b[0] = recAck
700- binary.BigEndian.PutUint64(b[1:], seq)
701- _ = t.sendPayload(b[:])
702-}
703-func (t *Transport) sendPayload(p []byte) error {
704- frags, err := pdgram.FragmentPayload(t.nextCounter(), p, t.mtu-pdgram.HeaderSize-t.codec.Overhead())
705- if err != nil {
706- return err
707- }
708- for _, f := range frags {
709- raw, err := pdgram.MarshalFragment(f)
710- if err != nil {
711- return err
712- }
713- pkt := t.codec.Seal(t.sendDir, t.nextCounter(), raw, nil)
714- t.mu.Lock()
715- peer := t.peer
716- closed := t.closed
717- closeErr := t.closeErr
718- t.mu.Unlock()
719- if closed {
720- if closeErr != nil {
721- return closeErr
722- }
723- return errors.New("dgram: closed")
724- }
725- if peer == nil {
726- return errors.New("dgram: no peer")
727- }
728- if err := t.writePacket(pkt, peer); err != nil {
729- return err
730- }
731- }
732- return nil
733-}
734-
735384 func (t *Transport) pendingExists(seq uint64) bool {
736385 t.mu.Lock()
737386 defer t.mu.Unlock()
@@ -739,46 +388,6 @@ func (t *Transport) pendingExists(seq uint64) bool {
739388 return ok
740389 }
741390
742-func (t *Transport) writePacket(pkt []byte, peer net.Addr) error {
743- t.writeMu.Lock()
744- defer t.writeMu.Unlock()
745-
746- t.mu.Lock()
747- pc := t.pc
748- writeTimeout := t.writeTimeout
749- writeDeadlines := t.writeDeadlines
750- t.mu.Unlock()
751-
752- return t.writeDatagram(pc, peer, pkt, writeDeadlines, writeTimeout)
753-}
754-
755-func (t *Transport) writeDatagram(pc net.PacketConn, peer net.Addr, pkt []byte, deadlines *writeDeadlineState, timeout time.Duration) error {
756- if timeout <= 0 {
757- _, err := pc.WriteTo(pkt, peer)
758- return err
759- }
760- ownDeadline := t.clock.Now().Add(timeout)
761- finish, err := deadlines.begin(ownDeadline)
762- if err != nil {
763- return err
764- }
765- defer finish()
766- for {
767- _, err = pc.WriteTo(pkt, peer)
768- if err == nil {
769- return nil
770- }
771- var timeoutErr interface{ Timeout() bool }
772- now := t.clock.Now()
773- if !errors.As(err, &timeoutErr) || !timeoutErr.Timeout() || !now.Before(ownDeadline) {
774- return err
775- }
776- deadlines.expire(now)
777- }
778-}
779-
780-func (t *Transport) nextCounter() uint64 { t.mu.Lock(); defer t.mu.Unlock(); t.ctr++; return t.ctr }
781-
782391 func (t *Transport) readLoop(pc net.PacketConn) {
783392 buf := make([]byte, 64*1024)
784393 for {
@@ -1567,40 +1176,6 @@ func diagnosticAge(now, then time.Time) time.Duration {
15671176 return now.Sub(then)
15681177 }
15691178
1570-func (t *Transport) closeWithError(err error) {
1571- t.mu.Lock()
1572- closed := t.closeWithErrorLocked(err)
1573- t.mu.Unlock()
1574- if closed {
1575- t.broadcastClosed()
1576- }
1577-}
1578-
1579-func (t *Transport) closeWithErrorLocked(err error) bool {
1580- if t.closed {
1581- return false
1582- }
1583- t.closed = true
1584- t.closeErr = err
1585- for _, q := range t.outputQueue {
1586- if q.done != nil {
1587- q.done <- err
1588- close(q.done)
1589- }
1590- }
1591- t.outputQueue = nil
1592- t.probeWait = make(map[uint64]chan struct{})
1593- close(t.done)
1594- t.notifySendWaitersLocked()
1595- return true
1596-}
1597-
1598-func (t *Transport) broadcastClosed() {
1599- t.deliverMu.Lock()
1600- t.deliverCond.Broadcast()
1601- t.deliverMu.Unlock()
1602-}
1603-
16041179 func encodeData(seq uint64, reliable bool, f ports.Frame) []byte {
16051180 b := make([]byte, dataRecordHeaderSize+len(f.Payload))
16061181 b[0] = recData
@@ -324,319 +324,6 @@ func (s *writeDeadlineState) expire(now time.Time) {
324 s.mu.Unlock()324 s.mu.Unlock()
325 }325 }
326 326
327-func NewTransport(pc net.PacketConn, peer net.Addr, key []byte, sendDir, recvDir uint32, runtimeOpts ...RuntimeOption) (*Transport, error) {
328- return NewTransportWithOptions(pc, peer, key, sendDir, recvDir, Options{}, runtimeOpts...)
329-}
330-
331-func NewTransportWithOptions(pc net.PacketConn, peer net.Addr, key []byte, sendDir, recvDir uint32, opts Options, runtimeOpts ...RuntimeOption) (*Transport, error) {
332- c, err := pdgram.NewCodec(key)
333- if err != nil {
334- return nil, err
335- }
336- opts = normalizeOptions(opts)
337- t := &Transport{
338- pc: pc,
339- codec: c,
340- sendDir: sendDir,
341- recvDir: recvDir,
342- mtu: opts.MTU,
343- peer: peer,
344- pending: make(map[uint64]*pending),
345- replay: pdgram.NewReplayWindow(),
346- reasm: pdgram.NewReassembler(),
347- in: make(chan ports.Frame, 32),
348- done: make(chan struct{}),
349- health: newHealthTracker(opts.Clock.Now()),
350- lastAuthenticatedPacket: opts.Clock.Now(),
351- lastCompleteRecord: opts.Clock.Now(),
352- lastACKProgress: opts.Clock.Now(),
353- observe: opts.Observe,
354- heartbeat: opts.Heartbeat,
355- resendAfter: opts.ResendAfter,
356- maxResendAfter: opts.MaxResendAfter,
357- maxResendPerTick: opts.MaxResendPerTick,
358- rto: opts.ResendAfter,
359- writeTimeout: opts.WriteTimeout,
360- degradedAfter: opts.DegradedAfter,
361- probeAfter: opts.ProbeAfter,
362- offlineAfter: opts.OfflineAfter,
363- deadAfter: opts.DeadAfter,
364- maxPending: opts.MaxPending,
365- maxPendingWait: opts.MaxPendingWait,
366- maxRecvBuffer: opts.MaxRecvBuffer,
367- clock: opts.Clock,
368- congestion: newCongestionController(opts.MTU),
369- linkState: ports.LinkStateConnected,
370- linkEvents: make(chan ports.LinkEvent, linkEventBufferSize),
371- probeWait: make(map[uint64]chan struct{}),
372- rebind: opts.RebindPacketConn,
373- nextRecvSeq: 1,
374- recvBuf: make(map[uint64]ports.Frame),
375- }
376- for _, opt := range runtimeOpts {
377- if opt != nil {
378- opt(t)
379- }
380- }
381- t.sendWake = make(chan struct{})
382- t.control = make(chan controlRecord, controlQueueSize)
383- t.ackWake = make(chan struct{}, 1)
384- t.ackSend = make(chan uint64, 1)
385- t.outputWake = make(chan struct{})
386- t.dataSend = make(chan dataSendJob, opts.MaxPending)
387- t.writeDeadlines = newWriteDeadlineState(pc)
388- // One queued batch bounds retransmit work while its sole sender is paced.
389- t.retransmitWork = make(chan []retransmitRecord, 1)
390- if t.observe != nil {
391- t.diagnosticCh = make(chan Diagnostic, diagnosticBufferSize)
392- go t.diagnosticLoop()
393- }
394- t.deliverCond = sync.NewCond(&t.deliverMu)
395- go t.readLoop(pc)
396- go t.retransmitLoop()
397- go t.heartbeatLoop()
398- go t.healthLoop()
399- go t.dataSendLoop()
400- go t.deliveryLoop()
401- go t.controlLoop()
402- go t.ackSendLoop()
403- go t.outputPaceLoop()
404- t.emitDiagnostic()
405- return t, nil
406-}
407-
408-func (t *Transport) Send(f ports.Frame) error {
409- end := t.beginRuntimeOperation(ports.RuntimeAdapterSendStart, uint64(len(f.Payload)))
410- err := t.send(f, false)
411- end(err == nil)
412- return err
413-}
414-
415-func (t *Transport) SendAsync(f ports.Frame) error {
416- end := t.beginRuntimeOperation(ports.RuntimeAdapterSendStart, uint64(len(f.Payload)))
417- err := t.send(f, true)
418- end(err == nil)
419- return err
420-}
421-
422-// SendSynchronous owns queue reservation, fragment pacing, per-write deadlines,
423-// and close cancellation. Callers must not layer a shorter preflight timer over it.
424-func (t *Transport) SendSynchronous(f ports.Frame) error { return t.Send(f) }
425-
426-func (t *Transport) send(f ports.Frame, async bool) error {
427- reliable := true
428- if err := t.lockOutboundSlot(reliable); err != nil {
429- return err
430- }
431- unlockOutbound := true
432- defer func() {
433- if unlockOutbound {
434- t.outboundMu.Unlock()
435- }
436- }()
437- // lockOutboundSlot returns with both outboundMu and mu held.
438-
439- t.seq++
440- seq := t.seq
441- if reliable {
442- now := t.clock.Now()
443- if len(t.pending) == 0 {
444- t.health.pendingStarted(now)
445- }
446- t.pending[seq] = &pending{frame: f, enqueued: now}
447- }
448- var done chan error
449- if !async {
450- done = make(chan error, 1)
451- }
452- job := dataSendJob{seq: seq, reliable: reliable, frame: f, done: done}
453- if shouldPaceOutput(f) || len(t.outputQueue) > 0 {
454- wasEmpty := len(t.outputQueue) == 0
455- t.outputQueue = append(t.outputQueue, queuedSend(job))
456- if wasEmpty {
457- t.notifyOutputPacerLocked()
458- }
459- t.mu.Unlock()
460- t.outboundMu.Unlock()
461- unlockOutbound = false
462- if async {
463- return nil
464- }
465- return t.waitQueuedSend(done)
466- }
467- t.mu.Unlock()
468- if err := t.queueDataJob(job); err != nil {
469- t.removePending(seq, reliable)
470- return err
471- }
472- t.outboundMu.Unlock()
473- unlockOutbound = false
474- if async {
475- return nil
476- }
477- return t.waitQueuedSend(done)
478-}
479-
480-func (t *Transport) queueDataJob(job dataSendJob) error {
481- select {
482- case t.dataSend <- job:
483- return nil
484- case <-t.done:
485- return t.closedError()
486- }
487-}
488-
489-func (t *Transport) waitQueuedSend(done <-chan error) error {
490- select {
491- case err := <-done:
492- return err
493- case <-t.done:
494- t.mu.Lock()
495- err := t.closeErr
496- t.mu.Unlock()
497- if err != nil {
498- return err
499- }
500- return errors.New("dgram: closed")
501- }
502-}
503-
504-func (t *Transport) closedError() error {
505- t.mu.Lock()
506- defer t.mu.Unlock()
507- if t.closeErr != nil {
508- return t.closeErr
509- }
510- return errors.New("dgram: closed")
511-}
512-
513-func (t *Transport) lockOutboundSlot(reliable bool) error {
514- var pendingWaitStarted time.Time
515- for {
516- t.outboundMu.Lock()
517- t.mu.Lock()
518- if t.closed {
519- err := t.closeErr
520- t.mu.Unlock()
521- t.outboundMu.Unlock()
522- if err != nil {
523- return err
524- }
525- return errors.New("dgram: closed")
526- }
527- if !reliable || len(t.pending) < t.maxPending {
528- return nil
529- }
530- if t.linkState != ports.LinkStateConnected {
531- t.mu.Unlock()
532- t.outboundMu.Unlock()
533- return ErrPendingFull
534- }
535- if pendingWaitStarted.IsZero() {
536- pendingWaitStarted = t.clock.Now()
537- }
538- remaining := t.maxPendingWait - t.clock.Now().Sub(pendingWaitStarted)
539- if remaining <= 0 {
540- t.mu.Unlock()
541- t.outboundMu.Unlock()
542- return ErrPendingFull
543- }
544- wake := t.sendWake
545- timer := t.clock.NewTimer(remaining)
546- t.mu.Unlock()
547- t.outboundMu.Unlock()
548- select {
549- case <-timer.C():
550- case <-wake:
551- case <-t.done:
552- }
553- timer.Stop()
554- }
555-}
556-
557-func (t *Transport) removePending(seq uint64, reliable bool) {
558- if !reliable {
559- return
560- }
561- t.mu.Lock()
562- removed := t.removePendingLocked(seq)
563- if removed {
564- t.notifySendWaitersLocked()
565- }
566- t.mu.Unlock()
567-}
568-
569-func (t *Transport) removeQueuedPending(queued []queuedSend, err error) {
570- t.mu.Lock()
571- removed := false
572- for _, q := range queued {
573- if q.reliable {
574- removed = t.removePendingLocked(q.seq) || removed
575- }
576- if q.done != nil {
577- q.done <- err
578- close(q.done)
579- }
580- }
581- if removed {
582- t.notifySendWaitersLocked()
583- }
584- t.mu.Unlock()
585-}
586-
587-func (t *Transport) removePendingLocked(seq uint64) bool {
588- if _, ok := t.pending[seq]; !ok {
589- return false
590- }
591- delete(t.pending, seq)
592- if len(t.pending) == 0 {
593- t.health.pendingCleared()
594- }
595- return true
596-}
597-
598-func (t *Transport) Recv() (ports.Frame, error) {
599- end := t.beginRuntimeOperation(ports.RuntimeAdapterReceiveStart, 0)
600- select {
601- case f := <-t.in:
602- end(true)
603- return f, nil
604- case <-t.done:
605- t.mu.Lock()
606- err := t.closeErr
607- t.mu.Unlock()
608- if err != nil {
609- end(false)
610- return ports.Frame{}, err
611- }
612- err = errors.New("dgram: closed")
613- end(false)
614- return ports.Frame{}, err
615- }
616-}
617-func (t *Transport) Close() error {
618- done := t.beginShutdown()
619- t.closeWithError(errors.New("dgram: closed"))
620- t.mu.Lock()
621- pc := t.pc
622- t.mu.Unlock()
623- err := pc.Close()
624- if done != nil {
625- <-done
626- }
627- return err
628-}
629-
630-func (t *Transport) Peer() net.Addr { t.mu.Lock(); defer t.mu.Unlock(); return t.peer }
631-
632-func (t *Transport) LinkState() ports.LinkState {
633- t.mu.Lock()
634- defer t.mu.Unlock()
635- return t.linkState
636-}
637-
638-func (t *Transport) LinkEvents() <-chan ports.LinkEvent { return t.linkEvents }
639-
640 // Probe sends an authenticated datagram and waits for the peer to authenticate a327 // Probe sends an authenticated datagram and waits for the peer to authenticate a
641 // response or for ctx to expire. It is intended for UDP bootstrap reachability checks.328 // response or for ctx to expire. It is intended for UDP bootstrap reachability checks.
642 func (t *Transport) Probe(ctx context.Context) error {329 func (t *Transport) Probe(ctx context.Context) error {
@@ -694,44 +381,6 @@ func (t *Transport) unregisterProbe(id uint64) {
694 t.mu.Unlock()381 t.mu.Unlock()
695 }382 }
696 383
697-func (t *Transport) sendAck(seq uint64) {
698- var b [9]byte
699- b[0] = recAck
700- binary.BigEndian.PutUint64(b[1:], seq)
701- _ = t.sendPayload(b[:])
702-}
703-func (t *Transport) sendPayload(p []byte) error {
704- frags, err := pdgram.FragmentPayload(t.nextCounter(), p, t.mtu-pdgram.HeaderSize-t.codec.Overhead())
705- if err != nil {
706- return err
707- }
708- for _, f := range frags {
709- raw, err := pdgram.MarshalFragment(f)
710- if err != nil {
711- return err
712- }
713- pkt := t.codec.Seal(t.sendDir, t.nextCounter(), raw, nil)
714- t.mu.Lock()
715- peer := t.peer
716- closed := t.closed
717- closeErr := t.closeErr
718- t.mu.Unlock()
719- if closed {
720- if closeErr != nil {
721- return closeErr
722- }
723- return errors.New("dgram: closed")
724- }
725- if peer == nil {
726- return errors.New("dgram: no peer")
727- }
728- if err := t.writePacket(pkt, peer); err != nil {
729- return err
730- }
731- }
732- return nil
733-}
734-
735 func (t *Transport) pendingExists(seq uint64) bool {384 func (t *Transport) pendingExists(seq uint64) bool {
736 t.mu.Lock()385 t.mu.Lock()
737 defer t.mu.Unlock()386 defer t.mu.Unlock()
@@ -739,46 +388,6 @@ func (t *Transport) pendingExists(seq uint64) bool {
739 return ok388 return ok
740 }389 }
741 390
742-func (t *Transport) writePacket(pkt []byte, peer net.Addr) error {
743- t.writeMu.Lock()
744- defer t.writeMu.Unlock()
745-
746- t.mu.Lock()
747- pc := t.pc
748- writeTimeout := t.writeTimeout
749- writeDeadlines := t.writeDeadlines
750- t.mu.Unlock()
751-
752- return t.writeDatagram(pc, peer, pkt, writeDeadlines, writeTimeout)
753-}
754-
755-func (t *Transport) writeDatagram(pc net.PacketConn, peer net.Addr, pkt []byte, deadlines *writeDeadlineState, timeout time.Duration) error {
756- if timeout <= 0 {
757- _, err := pc.WriteTo(pkt, peer)
758- return err
759- }
760- ownDeadline := t.clock.Now().Add(timeout)
761- finish, err := deadlines.begin(ownDeadline)
762- if err != nil {
763- return err
764- }
765- defer finish()
766- for {
767- _, err = pc.WriteTo(pkt, peer)
768- if err == nil {
769- return nil
770- }
771- var timeoutErr interface{ Timeout() bool }
772- now := t.clock.Now()
773- if !errors.As(err, &timeoutErr) || !timeoutErr.Timeout() || !now.Before(ownDeadline) {
774- return err
775- }
776- deadlines.expire(now)
777- }
778-}
779-
780-func (t *Transport) nextCounter() uint64 { t.mu.Lock(); defer t.mu.Unlock(); t.ctr++; return t.ctr }
781-
782 func (t *Transport) readLoop(pc net.PacketConn) {391 func (t *Transport) readLoop(pc net.PacketConn) {
783 buf := make([]byte, 64*1024)392 buf := make([]byte, 64*1024)
784 for {393 for {
@@ -1567,40 +1176,6 @@ func diagnosticAge(now, then time.Time) time.Duration {
1567 return now.Sub(then)1176 return now.Sub(then)
1568 }1177 }
1569 1178
1570-func (t *Transport) closeWithError(err error) {
1571- t.mu.Lock()
1572- closed := t.closeWithErrorLocked(err)
1573- t.mu.Unlock()
1574- if closed {
1575- t.broadcastClosed()
1576- }
1577-}
1578-
1579-func (t *Transport) closeWithErrorLocked(err error) bool {
1580- if t.closed {
1581- return false
1582- }
1583- t.closed = true
1584- t.closeErr = err
1585- for _, q := range t.outputQueue {
1586- if q.done != nil {
1587- q.done <- err
1588- close(q.done)
1589- }
1590- }
1591- t.outputQueue = nil
1592- t.probeWait = make(map[uint64]chan struct{})
1593- close(t.done)
1594- t.notifySendWaitersLocked()
1595- return true
1596-}
1597-
1598-func (t *Transport) broadcastClosed() {
1599- t.deliverMu.Lock()
1600- t.deliverCond.Broadcast()
1601- t.deliverMu.Unlock()
1602-}
1603-
1604 func encodeData(seq uint64, reliable bool, f ports.Frame) []byte {1179 func encodeData(seq uint64, reliable bool, f ports.Frame) []byte {
1605 b := make([]byte, dataRecordHeaderSize+len(f.Payload))1180 b := make([]byte, dataRecordHeaderSize+len(f.Payload))
1606 b[0] = recData1181 b[0] = recData
A internal/adapters/dgram/transport_lifecycle.go +177 -0
@@ -0,0 +1,177 @@
1+package dgram
2+
3+import (
4+ "errors"
5+ "net"
6+ "sync"
7+
8+ "github.com/bnema/vev/internal/ports"
9+ pdgram "github.com/bnema/vev/pkg/dgram"
10+)
11+
12+func NewTransport(pc net.PacketConn, peer net.Addr, key []byte, sendDir, recvDir uint32, runtimeOpts ...RuntimeOption) (*Transport, error) {
13+ return NewTransportWithOptions(pc, peer, key, sendDir, recvDir, Options{}, runtimeOpts...)
14+}
15+
16+func NewTransportWithOptions(pc net.PacketConn, peer net.Addr, key []byte, sendDir, recvDir uint32, opts Options, runtimeOpts ...RuntimeOption) (*Transport, error) {
17+ c, err := pdgram.NewCodec(key)
18+ if err != nil {
19+ return nil, err
20+ }
21+ opts = normalizeOptions(opts)
22+ t := &Transport{
23+ pc: pc,
24+ codec: c,
25+ sendDir: sendDir,
26+ recvDir: recvDir,
27+ mtu: opts.MTU,
28+ peer: peer,
29+ pending: make(map[uint64]*pending),
30+ replay: pdgram.NewReplayWindow(),
31+ reasm: pdgram.NewReassembler(),
32+ in: make(chan ports.Frame, 32),
33+ done: make(chan struct{}),
34+ health: newHealthTracker(opts.Clock.Now()),
35+ lastAuthenticatedPacket: opts.Clock.Now(),
36+ lastCompleteRecord: opts.Clock.Now(),
37+ lastACKProgress: opts.Clock.Now(),
38+ observe: opts.Observe,
39+ heartbeat: opts.Heartbeat,
40+ resendAfter: opts.ResendAfter,
41+ maxResendAfter: opts.MaxResendAfter,
42+ maxResendPerTick: opts.MaxResendPerTick,
43+ rto: opts.ResendAfter,
44+ writeTimeout: opts.WriteTimeout,
45+ degradedAfter: opts.DegradedAfter,
46+ probeAfter: opts.ProbeAfter,
47+ offlineAfter: opts.OfflineAfter,
48+ deadAfter: opts.DeadAfter,
49+ maxPending: opts.MaxPending,
50+ maxPendingWait: opts.MaxPendingWait,
51+ maxRecvBuffer: opts.MaxRecvBuffer,
52+ clock: opts.Clock,
53+ congestion: newCongestionController(opts.MTU),
54+ linkState: ports.LinkStateConnected,
55+ linkEvents: make(chan ports.LinkEvent, linkEventBufferSize),
56+ probeWait: make(map[uint64]chan struct{}),
57+ rebind: opts.RebindPacketConn,
58+ nextRecvSeq: 1,
59+ recvBuf: make(map[uint64]ports.Frame),
60+ }
61+ for _, opt := range runtimeOpts {
62+ if opt != nil {
63+ opt(t)
64+ }
65+ }
66+ t.sendWake = make(chan struct{})
67+ t.control = make(chan controlRecord, controlQueueSize)
68+ t.ackWake = make(chan struct{}, 1)
69+ t.ackSend = make(chan uint64, 1)
70+ t.outputWake = make(chan struct{})
71+ t.dataSend = make(chan dataSendJob, opts.MaxPending)
72+ t.writeDeadlines = newWriteDeadlineState(pc)
73+ // One queued batch bounds retransmit work while its sole sender is paced.
74+ t.retransmitWork = make(chan []retransmitRecord, 1)
75+ if t.observe != nil {
76+ t.diagnosticCh = make(chan Diagnostic, diagnosticBufferSize)
77+ go t.diagnosticLoop()
78+ }
79+ t.deliverCond = sync.NewCond(&t.deliverMu)
80+ go t.readLoop(pc)
81+ go t.retransmitLoop()
82+ go t.heartbeatLoop()
83+ go t.healthLoop()
84+ go t.dataSendLoop()
85+ go t.deliveryLoop()
86+ go t.controlLoop()
87+ go t.ackSendLoop()
88+ go t.outputPaceLoop()
89+ t.emitDiagnostic()
90+ return t, nil
91+}
92+
93+func (t *Transport) Recv() (ports.Frame, error) {
94+ end := t.beginRuntimeOperation(ports.RuntimeAdapterReceiveStart, 0)
95+ select {
96+ case f := <-t.in:
97+ end(true)
98+ return f, nil
99+ case <-t.done:
100+ t.mu.Lock()
101+ err := t.closeErr
102+ t.mu.Unlock()
103+ if err != nil {
104+ end(false)
105+ return ports.Frame{}, err
106+ }
107+ err = errors.New("dgram: closed")
108+ end(false)
109+ return ports.Frame{}, err
110+ }
111+}
112+
113+func (t *Transport) Close() error {
114+ done := t.beginShutdown()
115+ t.closeWithError(errors.New("dgram: closed"))
116+ t.mu.Lock()
117+ pc := t.pc
118+ t.mu.Unlock()
119+ err := pc.Close()
120+ if done != nil {
121+ <-done
122+ }
123+ return err
124+}
125+
126+func (t *Transport) Peer() net.Addr { t.mu.Lock(); defer t.mu.Unlock(); return t.peer }
127+
128+func (t *Transport) LinkState() ports.LinkState {
129+ t.mu.Lock()
130+ defer t.mu.Unlock()
131+ return t.linkState
132+}
133+
134+func (t *Transport) LinkEvents() <-chan ports.LinkEvent { return t.linkEvents }
135+
136+func (t *Transport) closedError() error {
137+ t.mu.Lock()
138+ defer t.mu.Unlock()
139+ if t.closeErr != nil {
140+ return t.closeErr
141+ }
142+ return errors.New("dgram: closed")
143+}
144+
145+func (t *Transport) closeWithError(err error) {
146+ t.mu.Lock()
147+ closed := t.closeWithErrorLocked(err)
148+ t.mu.Unlock()
149+ if closed {
150+ t.broadcastClosed()
151+ }
152+}
153+
154+func (t *Transport) closeWithErrorLocked(err error) bool {
155+ if t.closed {
156+ return false
157+ }
158+ t.closed = true
159+ t.closeErr = err
160+ for _, q := range t.outputQueue {
161+ if q.done != nil {
162+ q.done <- err
163+ close(q.done)
164+ }
165+ }
166+ t.outputQueue = nil
167+ t.probeWait = make(map[uint64]chan struct{})
168+ close(t.done)
169+ t.notifySendWaitersLocked()
170+ return true
171+}
172+
173+func (t *Transport) broadcastClosed() {
174+ t.deliverMu.Lock()
175+ t.deliverCond.Broadcast()
176+ t.deliverMu.Unlock()
177+}
@@ -0,0 +1,177 @@
1+package dgram
2+
3+import (
4+ "errors"
5+ "net"
6+ "sync"
7+
8+ "github.com/bnema/vev/internal/ports"
9+ pdgram "github.com/bnema/vev/pkg/dgram"
10+)
11+
12+func NewTransport(pc net.PacketConn, peer net.Addr, key []byte, sendDir, recvDir uint32, runtimeOpts ...RuntimeOption) (*Transport, error) {
13+ return NewTransportWithOptions(pc, peer, key, sendDir, recvDir, Options{}, runtimeOpts...)
14+}
15+
16+func NewTransportWithOptions(pc net.PacketConn, peer net.Addr, key []byte, sendDir, recvDir uint32, opts Options, runtimeOpts ...RuntimeOption) (*Transport, error) {
17+ c, err := pdgram.NewCodec(key)
18+ if err != nil {
19+ return nil, err
20+ }
21+ opts = normalizeOptions(opts)
22+ t := &Transport{
23+ pc: pc,
24+ codec: c,
25+ sendDir: sendDir,
26+ recvDir: recvDir,
27+ mtu: opts.MTU,
28+ peer: peer,
29+ pending: make(map[uint64]*pending),
30+ replay: pdgram.NewReplayWindow(),
31+ reasm: pdgram.NewReassembler(),
32+ in: make(chan ports.Frame, 32),
33+ done: make(chan struct{}),
34+ health: newHealthTracker(opts.Clock.Now()),
35+ lastAuthenticatedPacket: opts.Clock.Now(),
36+ lastCompleteRecord: opts.Clock.Now(),
37+ lastACKProgress: opts.Clock.Now(),
38+ observe: opts.Observe,
39+ heartbeat: opts.Heartbeat,
40+ resendAfter: opts.ResendAfter,
41+ maxResendAfter: opts.MaxResendAfter,
42+ maxResendPerTick: opts.MaxResendPerTick,
43+ rto: opts.ResendAfter,
44+ writeTimeout: opts.WriteTimeout,
45+ degradedAfter: opts.DegradedAfter,
46+ probeAfter: opts.ProbeAfter,
47+ offlineAfter: opts.OfflineAfter,
48+ deadAfter: opts.DeadAfter,
49+ maxPending: opts.MaxPending,
50+ maxPendingWait: opts.MaxPendingWait,
51+ maxRecvBuffer: opts.MaxRecvBuffer,
52+ clock: opts.Clock,
53+ congestion: newCongestionController(opts.MTU),
54+ linkState: ports.LinkStateConnected,
55+ linkEvents: make(chan ports.LinkEvent, linkEventBufferSize),
56+ probeWait: make(map[uint64]chan struct{}),
57+ rebind: opts.RebindPacketConn,
58+ nextRecvSeq: 1,
59+ recvBuf: make(map[uint64]ports.Frame),
60+ }
61+ for _, opt := range runtimeOpts {
62+ if opt != nil {
63+ opt(t)
64+ }
65+ }
66+ t.sendWake = make(chan struct{})
67+ t.control = make(chan controlRecord, controlQueueSize)
68+ t.ackWake = make(chan struct{}, 1)
69+ t.ackSend = make(chan uint64, 1)
70+ t.outputWake = make(chan struct{})
71+ t.dataSend = make(chan dataSendJob, opts.MaxPending)
72+ t.writeDeadlines = newWriteDeadlineState(pc)
73+ // One queued batch bounds retransmit work while its sole sender is paced.
74+ t.retransmitWork = make(chan []retransmitRecord, 1)
75+ if t.observe != nil {
76+ t.diagnosticCh = make(chan Diagnostic, diagnosticBufferSize)
77+ go t.diagnosticLoop()
78+ }
79+ t.deliverCond = sync.NewCond(&t.deliverMu)
80+ go t.readLoop(pc)
81+ go t.retransmitLoop()
82+ go t.heartbeatLoop()
83+ go t.healthLoop()
84+ go t.dataSendLoop()
85+ go t.deliveryLoop()
86+ go t.controlLoop()
87+ go t.ackSendLoop()
88+ go t.outputPaceLoop()
89+ t.emitDiagnostic()
90+ return t, nil
91+}
92+
93+func (t *Transport) Recv() (ports.Frame, error) {
94+ end := t.beginRuntimeOperation(ports.RuntimeAdapterReceiveStart, 0)
95+ select {
96+ case f := <-t.in:
97+ end(true)
98+ return f, nil
99+ case <-t.done:
100+ t.mu.Lock()
101+ err := t.closeErr
102+ t.mu.Unlock()
103+ if err != nil {
104+ end(false)
105+ return ports.Frame{}, err
106+ }
107+ err = errors.New("dgram: closed")
108+ end(false)
109+ return ports.Frame{}, err
110+ }
111+}
112+
113+func (t *Transport) Close() error {
114+ done := t.beginShutdown()
115+ t.closeWithError(errors.New("dgram: closed"))
116+ t.mu.Lock()
117+ pc := t.pc
118+ t.mu.Unlock()
119+ err := pc.Close()
120+ if done != nil {
121+ <-done
122+ }
123+ return err
124+}
125+
126+func (t *Transport) Peer() net.Addr { t.mu.Lock(); defer t.mu.Unlock(); return t.peer }
127+
128+func (t *Transport) LinkState() ports.LinkState {
129+ t.mu.Lock()
130+ defer t.mu.Unlock()
131+ return t.linkState
132+}
133+
134+func (t *Transport) LinkEvents() <-chan ports.LinkEvent { return t.linkEvents }
135+
136+func (t *Transport) closedError() error {
137+ t.mu.Lock()
138+ defer t.mu.Unlock()
139+ if t.closeErr != nil {
140+ return t.closeErr
141+ }
142+ return errors.New("dgram: closed")
143+}
144+
145+func (t *Transport) closeWithError(err error) {
146+ t.mu.Lock()
147+ closed := t.closeWithErrorLocked(err)
148+ t.mu.Unlock()
149+ if closed {
150+ t.broadcastClosed()
151+ }
152+}
153+
154+func (t *Transport) closeWithErrorLocked(err error) bool {
155+ if t.closed {
156+ return false
157+ }
158+ t.closed = true
159+ t.closeErr = err
160+ for _, q := range t.outputQueue {
161+ if q.done != nil {
162+ q.done <- err
163+ close(q.done)
164+ }
165+ }
166+ t.outputQueue = nil
167+ t.probeWait = make(map[uint64]chan struct{})
168+ close(t.done)
169+ t.notifySendWaitersLocked()
170+ return true
171+}
172+
173+func (t *Transport) broadcastClosed() {
174+ t.deliverMu.Lock()
175+ t.deliverCond.Broadcast()
176+ t.deliverMu.Unlock()
177+}
A internal/adapters/dgram/transport_send.go +230 -0
@@ -0,0 +1,230 @@
1+package dgram
2+
3+import (
4+ "errors"
5+ "net"
6+ "time"
7+
8+ "github.com/bnema/vev/internal/ports"
9+)
10+
11+func (t *Transport) Send(f ports.Frame) error {
12+ end := t.beginRuntimeOperation(ports.RuntimeAdapterSendStart, uint64(len(f.Payload)))
13+ err := t.send(f, false)
14+ end(err == nil)
15+ return err
16+}
17+
18+func (t *Transport) SendAsync(f ports.Frame) error {
19+ end := t.beginRuntimeOperation(ports.RuntimeAdapterSendStart, uint64(len(f.Payload)))
20+ err := t.send(f, true)
21+ end(err == nil)
22+ return err
23+}
24+
25+// SendSynchronous owns queue reservation, fragment pacing, per-write deadlines,
26+// and close cancellation. Callers must not layer a shorter preflight timer over it.
27+func (t *Transport) SendSynchronous(f ports.Frame) error { return t.Send(f) }
28+
29+func (t *Transport) send(f ports.Frame, async bool) error {
30+ reliable := true
31+ if err := t.lockOutboundSlot(reliable); err != nil {
32+ return err
33+ }
34+ unlockOutbound := true
35+ defer func() {
36+ if unlockOutbound {
37+ t.outboundMu.Unlock()
38+ }
39+ }()
40+ // lockOutboundSlot returns with both outboundMu and mu held.
41+
42+ t.seq++
43+ seq := t.seq
44+ if reliable {
45+ now := t.clock.Now()
46+ if len(t.pending) == 0 {
47+ t.health.pendingStarted(now)
48+ }
49+ t.pending[seq] = &pending{frame: f, enqueued: now}
50+ }
51+ var done chan error
52+ if !async {
53+ done = make(chan error, 1)
54+ }
55+ job := dataSendJob{seq: seq, reliable: reliable, frame: f, done: done}
56+ if shouldPaceOutput(f) || len(t.outputQueue) > 0 {
57+ wasEmpty := len(t.outputQueue) == 0
58+ t.outputQueue = append(t.outputQueue, queuedSend(job))
59+ if wasEmpty {
60+ t.notifyOutputPacerLocked()
61+ }
62+ t.mu.Unlock()
63+ t.outboundMu.Unlock()
64+ unlockOutbound = false
65+ if async {
66+ return nil
67+ }
68+ return t.waitQueuedSend(done)
69+ }
70+ t.mu.Unlock()
71+ if err := t.queueDataJob(job); err != nil {
72+ t.removePending(seq, reliable)
73+ return err
74+ }
75+ t.outboundMu.Unlock()
76+ unlockOutbound = false
77+ if async {
78+ return nil
79+ }
80+ return t.waitQueuedSend(done)
81+}
82+
83+func (t *Transport) queueDataJob(job dataSendJob) error {
84+ select {
85+ case t.dataSend <- job:
86+ return nil
87+ case <-t.done:
88+ return t.closedError()
89+ }
90+}
91+
92+func (t *Transport) waitQueuedSend(done <-chan error) error {
93+ select {
94+ case err := <-done:
95+ return err
96+ case <-t.done:
97+ t.mu.Lock()
98+ err := t.closeErr
99+ t.mu.Unlock()
100+ if err != nil {
101+ return err
102+ }
103+ return errors.New("dgram: closed")
104+ }
105+}
106+
107+func (t *Transport) lockOutboundSlot(reliable bool) error {
108+ var pendingWaitStarted time.Time
109+ for {
110+ t.outboundMu.Lock()
111+ t.mu.Lock()
112+ if t.closed {
113+ err := t.closeErr
114+ t.mu.Unlock()
115+ t.outboundMu.Unlock()
116+ if err != nil {
117+ return err
118+ }
119+ return errors.New("dgram: closed")
120+ }
121+ if !reliable || len(t.pending) < t.maxPending {
122+ return nil
123+ }
124+ if t.linkState != ports.LinkStateConnected {
125+ t.mu.Unlock()
126+ t.outboundMu.Unlock()
127+ return ErrPendingFull
128+ }
129+ if pendingWaitStarted.IsZero() {
130+ pendingWaitStarted = t.clock.Now()
131+ }
132+ remaining := t.maxPendingWait - t.clock.Now().Sub(pendingWaitStarted)
133+ if remaining <= 0 {
134+ t.mu.Unlock()
135+ t.outboundMu.Unlock()
136+ return ErrPendingFull
137+ }
138+ wake := t.sendWake
139+ timer := t.clock.NewTimer(remaining)
140+ t.mu.Unlock()
141+ t.outboundMu.Unlock()
142+ select {
143+ case <-timer.C():
144+ case <-wake:
145+ case <-t.done:
146+ }
147+ timer.Stop()
148+ }
149+}
150+
151+func (t *Transport) removePending(seq uint64, reliable bool) {
152+ if !reliable {
153+ return
154+ }
155+ t.mu.Lock()
156+ removed := t.removePendingLocked(seq)
157+ if removed {
158+ t.notifySendWaitersLocked()
159+ }
160+ t.mu.Unlock()
161+}
162+
163+func (t *Transport) removeQueuedPending(queued []queuedSend, err error) {
164+ t.mu.Lock()
165+ removed := false
166+ for _, q := range queued {
167+ if q.reliable {
168+ removed = t.removePendingLocked(q.seq) || removed
169+ }
170+ if q.done != nil {
171+ q.done <- err
172+ close(q.done)
173+ }
174+ }
175+ if removed {
176+ t.notifySendWaitersLocked()
177+ }
178+ t.mu.Unlock()
179+}
180+
181+func (t *Transport) removePendingLocked(seq uint64) bool {
182+ if _, ok := t.pending[seq]; !ok {
183+ return false
184+ }
185+ delete(t.pending, seq)
186+ if len(t.pending) == 0 {
187+ t.health.pendingCleared()
188+ }
189+ return true
190+}
191+
192+func (t *Transport) writePacket(pkt []byte, peer net.Addr) error {
193+ t.writeMu.Lock()
194+ defer t.writeMu.Unlock()
195+
196+ t.mu.Lock()
197+ pc := t.pc
198+ writeTimeout := t.writeTimeout
199+ writeDeadlines := t.writeDeadlines
200+ t.mu.Unlock()
201+
202+ return t.writeDatagram(pc, peer, pkt, writeDeadlines, writeTimeout)
203+}
204+
205+func (t *Transport) writeDatagram(pc net.PacketConn, peer net.Addr, pkt []byte, deadlines *writeDeadlineState, timeout time.Duration) error {
206+ if timeout <= 0 {
207+ _, err := pc.WriteTo(pkt, peer)
208+ return err
209+ }
210+ ownDeadline := t.clock.Now().Add(timeout)
211+ finish, err := deadlines.begin(ownDeadline)
212+ if err != nil {
213+ return err
214+ }
215+ defer finish()
216+ for {
217+ _, err = pc.WriteTo(pkt, peer)
218+ if err == nil {
219+ return nil
220+ }
221+ var timeoutErr interface{ Timeout() bool }
222+ now := t.clock.Now()
223+ if !errors.As(err, &timeoutErr) || !timeoutErr.Timeout() || !now.Before(ownDeadline) {
224+ return err
225+ }
226+ deadlines.expire(now)
227+ }
228+}
229+
230+func (t *Transport) nextCounter() uint64 { t.mu.Lock(); defer t.mu.Unlock(); t.ctr++; return t.ctr }
@@ -0,0 +1,230 @@
1+package dgram
2+
3+import (
4+ "errors"
5+ "net"
6+ "time"
7+
8+ "github.com/bnema/vev/internal/ports"
9+)
10+
11+func (t *Transport) Send(f ports.Frame) error {
12+ end := t.beginRuntimeOperation(ports.RuntimeAdapterSendStart, uint64(len(f.Payload)))
13+ err := t.send(f, false)
14+ end(err == nil)
15+ return err
16+}
17+
18+func (t *Transport) SendAsync(f ports.Frame) error {
19+ end := t.beginRuntimeOperation(ports.RuntimeAdapterSendStart, uint64(len(f.Payload)))
20+ err := t.send(f, true)
21+ end(err == nil)
22+ return err
23+}
24+
25+// SendSynchronous owns queue reservation, fragment pacing, per-write deadlines,
26+// and close cancellation. Callers must not layer a shorter preflight timer over it.
27+func (t *Transport) SendSynchronous(f ports.Frame) error { return t.Send(f) }
28+
29+func (t *Transport) send(f ports.Frame, async bool) error {
30+ reliable := true
31+ if err := t.lockOutboundSlot(reliable); err != nil {
32+ return err
33+ }
34+ unlockOutbound := true
35+ defer func() {
36+ if unlockOutbound {
37+ t.outboundMu.Unlock()
38+ }
39+ }()
40+ // lockOutboundSlot returns with both outboundMu and mu held.
41+
42+ t.seq++
43+ seq := t.seq
44+ if reliable {
45+ now := t.clock.Now()
46+ if len(t.pending) == 0 {
47+ t.health.pendingStarted(now)
48+ }
49+ t.pending[seq] = &pending{frame: f, enqueued: now}
50+ }
51+ var done chan error
52+ if !async {
53+ done = make(chan error, 1)
54+ }
55+ job := dataSendJob{seq: seq, reliable: reliable, frame: f, done: done}
56+ if shouldPaceOutput(f) || len(t.outputQueue) > 0 {
57+ wasEmpty := len(t.outputQueue) == 0
58+ t.outputQueue = append(t.outputQueue, queuedSend(job))
59+ if wasEmpty {
60+ t.notifyOutputPacerLocked()
61+ }
62+ t.mu.Unlock()
63+ t.outboundMu.Unlock()
64+ unlockOutbound = false
65+ if async {
66+ return nil
67+ }
68+ return t.waitQueuedSend(done)
69+ }
70+ t.mu.Unlock()
71+ if err := t.queueDataJob(job); err != nil {
72+ t.removePending(seq, reliable)
73+ return err
74+ }
75+ t.outboundMu.Unlock()
76+ unlockOutbound = false
77+ if async {
78+ return nil
79+ }
80+ return t.waitQueuedSend(done)
81+}
82+
83+func (t *Transport) queueDataJob(job dataSendJob) error {
84+ select {
85+ case t.dataSend <- job:
86+ return nil
87+ case <-t.done:
88+ return t.closedError()
89+ }
90+}
91+
92+func (t *Transport) waitQueuedSend(done <-chan error) error {
93+ select {
94+ case err := <-done:
95+ return err
96+ case <-t.done:
97+ t.mu.Lock()
98+ err := t.closeErr
99+ t.mu.Unlock()
100+ if err != nil {
101+ return err
102+ }
103+ return errors.New("dgram: closed")
104+ }
105+}
106+
107+func (t *Transport) lockOutboundSlot(reliable bool) error {
108+ var pendingWaitStarted time.Time
109+ for {
110+ t.outboundMu.Lock()
111+ t.mu.Lock()
112+ if t.closed {
113+ err := t.closeErr
114+ t.mu.Unlock()
115+ t.outboundMu.Unlock()
116+ if err != nil {
117+ return err
118+ }
119+ return errors.New("dgram: closed")
120+ }
121+ if !reliable || len(t.pending) < t.maxPending {
122+ return nil
123+ }
124+ if t.linkState != ports.LinkStateConnected {
125+ t.mu.Unlock()
126+ t.outboundMu.Unlock()
127+ return ErrPendingFull
128+ }
129+ if pendingWaitStarted.IsZero() {
130+ pendingWaitStarted = t.clock.Now()
131+ }
132+ remaining := t.maxPendingWait - t.clock.Now().Sub(pendingWaitStarted)
133+ if remaining <= 0 {
134+ t.mu.Unlock()
135+ t.outboundMu.Unlock()
136+ return ErrPendingFull
137+ }
138+ wake := t.sendWake
139+ timer := t.clock.NewTimer(remaining)
140+ t.mu.Unlock()
141+ t.outboundMu.Unlock()
142+ select {
143+ case <-timer.C():
144+ case <-wake:
145+ case <-t.done:
146+ }
147+ timer.Stop()
148+ }
149+}
150+
151+func (t *Transport) removePending(seq uint64, reliable bool) {
152+ if !reliable {
153+ return
154+ }
155+ t.mu.Lock()
156+ removed := t.removePendingLocked(seq)
157+ if removed {
158+ t.notifySendWaitersLocked()
159+ }
160+ t.mu.Unlock()
161+}
162+
163+func (t *Transport) removeQueuedPending(queued []queuedSend, err error) {
164+ t.mu.Lock()
165+ removed := false
166+ for _, q := range queued {
167+ if q.reliable {
168+ removed = t.removePendingLocked(q.seq) || removed
169+ }
170+ if q.done != nil {
171+ q.done <- err
172+ close(q.done)
173+ }
174+ }
175+ if removed {
176+ t.notifySendWaitersLocked()
177+ }
178+ t.mu.Unlock()
179+}
180+
181+func (t *Transport) removePendingLocked(seq uint64) bool {
182+ if _, ok := t.pending[seq]; !ok {
183+ return false
184+ }
185+ delete(t.pending, seq)
186+ if len(t.pending) == 0 {
187+ t.health.pendingCleared()
188+ }
189+ return true
190+}
191+
192+func (t *Transport) writePacket(pkt []byte, peer net.Addr) error {
193+ t.writeMu.Lock()
194+ defer t.writeMu.Unlock()
195+
196+ t.mu.Lock()
197+ pc := t.pc
198+ writeTimeout := t.writeTimeout
199+ writeDeadlines := t.writeDeadlines
200+ t.mu.Unlock()
201+
202+ return t.writeDatagram(pc, peer, pkt, writeDeadlines, writeTimeout)
203+}
204+
205+func (t *Transport) writeDatagram(pc net.PacketConn, peer net.Addr, pkt []byte, deadlines *writeDeadlineState, timeout time.Duration) error {
206+ if timeout <= 0 {
207+ _, err := pc.WriteTo(pkt, peer)
208+ return err
209+ }
210+ ownDeadline := t.clock.Now().Add(timeout)
211+ finish, err := deadlines.begin(ownDeadline)
212+ if err != nil {
213+ return err
214+ }
215+ defer finish()
216+ for {
217+ _, err = pc.WriteTo(pkt, peer)
218+ if err == nil {
219+ return nil
220+ }
221+ var timeoutErr interface{ Timeout() bool }
222+ now := t.clock.Now()
223+ if !errors.As(err, &timeoutErr) || !timeoutErr.Timeout() || !now.Before(ownDeadline) {
224+ return err
225+ }
226+ deadlines.expire(now)
227+ }
228+}
229+
230+func (t *Transport) nextCounter() uint64 { t.mu.Lock(); defer t.mu.Unlock(); t.ctr++; return t.ctr }

Keyboard shortcuts

Single-key shortcuts are on. ctrl k and esc work whatever this says.

ctrl k :
go to a section, branch, tag, file or repository
?
this list
j k
move the row cursor down / up
enter
open the row under the cursor; on Files, a directory opens or closes in place
↑ ↓ ← → home end
in the file tree (focus on a row): previous / next row, close / open a directory or go to its parent / first entry, first / last row. They never open a file
l h
on Files: open a directory or a file / close a directory or go up to its parent, then to the parent page. Elsewhere: into the side panel (on a merge request, its files) / back to the content; there j k move between its controls
g g G
first / last row
/
filter the list; on Files, find a file anywhere in the repository (esc clears, enter opens the best match)
b
switch branch or tag
g s g f g c g b g m g i g r g a
tabs: summary, files, commits, branches & tags, merge requests, issues, releases, actions (when available)
g h
home
w
wide mode: use the whole width of the screen, or not (remembered)
y
copy a permalink pinned to the commit
d u x
diff: split / unified / collapse or expand all
e E
diff: the whole file in view / every file, with the lines the hunks hide
n
merge request: next note in the diff
esc
close a dialog, leave a field

Branches 12

Tags 11

↑↓ moveenter openesc close