detached 3b6ed69

refactor(dgram): split transport receive and retransmit paths

commit
3b6ed69e9b1c26f1ddfb6f874356d4bed149dc21
author
brice · · 2026-07-14 20:47 UTC
parent
0f188ef
signature
GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email
4 files · +575 -558
Unified
UnifiedSplit
M internal/adapters/dgram/transport.go +0 -558
@@ -3,10 +3,8 @@ package dgram
33
44 import (
55 "context"
6- "encoding/binary"
76 "errors"
87 "net"
9- "slices"
108 "sync"
119 "time"
1210
@@ -388,263 +386,6 @@ func (t *Transport) pendingExists(seq uint64) bool {
388386 return ok
389387 }
390388
391-func (t *Transport) readLoop(pc net.PacketConn) {
392- buf := make([]byte, 64*1024)
393- for {
394- n, addr, err := pc.ReadFrom(buf)
395- if err != nil {
396- t.mu.Lock()
397- current := t.pc == pc
398- closed := t.closed
399- t.mu.Unlock()
400- if current && !closed {
401- t.closeWithError(err)
402- }
403- return
404- }
405- t.recvMu.Lock()
406- counter, pt, err := t.codec.Open(append([]byte(nil), buf[:n]...), t.recvDir, nil, t.replay)
407- if err != nil {
408- t.recvMu.Unlock()
409- continue
410- }
411- frag, err := pdgram.UnmarshalFragment(pt)
412- if err != nil {
413- t.recvMu.Unlock()
414- t.notifyMalformedFragment()
415- continue
416- }
417- payload, complete, err := t.reasm.Add(frag)
418- inflight := t.reasm.Inflight()
419- // Keep the diagnostic mirror ordered with reassembly mutations. Rebind can
420- // overlap old and new read loops, so recvMu must remain held through commit.
421- t.mu.Lock()
422- t.reassemblyInflight = inflight
423- var afterAuthenticated func()
424- if err == nil {
425- now := t.clock.Now()
426- t.health.authenticatedPacket(now)
427- t.lastAuthenticatedPacket = now
428- if !t.peerCounterSet || counter > t.peerCounter {
429- t.peer = addr
430- t.peerCounter = counter
431- t.peerCounterSet = true
432- }
433- afterAuthenticated = t.afterAuthenticatedPacket
434- }
435- t.mu.Unlock()
436- t.recvMu.Unlock()
437- if afterAuthenticated != nil {
438- afterAuthenticated()
439- }
440- if err != nil {
441- t.notifyMalformedFragment()
442- t.notifyPacketProcessed()
443- continue
444- }
445- if !complete {
446- t.notifyPacketProcessed()
447- continue
448- }
449- t.recordCompleteRecord()
450- t.handleRecord(payload)
451- t.checkSilence()
452- t.notifyPacketProcessed()
453- }
454-}
455-
456-func (t *Transport) notifyPacketProcessed() {
457- t.mu.Lock()
458- afterPacketProcessed := t.afterPacketProcessed
459- t.mu.Unlock()
460- if afterPacketProcessed != nil {
461- afterPacketProcessed()
462- }
463-}
464-
465-func (t *Transport) notifyMalformedFragment() {
466- t.mu.Lock()
467- afterMalformed := t.afterMalformedFragment
468- t.mu.Unlock()
469- if afterMalformed != nil {
470- afterMalformed()
471- }
472-}
473-
474-func (t *Transport) recordCompleteRecord() {
475- t.mu.Lock()
476- now := t.clock.Now()
477- t.health.completeRecord(now)
478- t.lastCompleteRecord = now
479- t.mu.Unlock()
480-}
481-
482-func (t *Transport) handleRecord(p []byte) {
483- if len(p) < 1 {
484- return
485- }
486- switch p[0] {
487- case recAck:
488- if len(p) != 9 {
489- return
490- }
491- seq := binary.BigEndian.Uint64(p[1:])
492- t.mu.Lock()
493- acked := false
494- ackedBytes := 0
495- for pendingSeq, p := range t.pending {
496- if pendingSeq > seq {
497- continue
498- }
499- if pendingSeq == seq && p != nil && !p.retransmitted {
500- t.updateRTTLocked(t.clock.Now().Sub(p.first))
501- }
502- if p != nil {
503- ackedBytes += p.wireBytes
504- }
505- delete(t.pending, pendingSeq)
506- acked = true
507- }
508- if acked {
509- t.congestion.onACK(ackedBytes)
510- now := t.clock.Now()
511- t.health.ackProgress(now)
512- if len(t.pending) == 0 {
513- t.health.pendingCleared()
514- }
515- t.lastACKProgress = now
516- t.notifySendWaitersLocked()
517- }
518- t.mu.Unlock()
519- if acked {
520- t.setLinkState(ports.LinkStateConnected, nil)
521- t.emitDiagnostic()
522- }
523- case recProbe:
524- if len(p) != 9 {
525- return
526- }
527- t.queueControl(recPong, binary.BigEndian.Uint64(p[1:]))
528- case recPong:
529- if len(p) != 9 {
530- return
531- }
532- id := binary.BigEndian.Uint64(p[1:])
533- t.mu.Lock()
534- ch := t.probeWait[id]
535- if ch != nil {
536- delete(t.probeWait, id)
537- close(ch)
538- }
539- t.mu.Unlock()
540- case recData:
541- seq, reliable, f, ok := decodeData(p)
542- if !ok {
543- return
544- }
545- if reliable {
546- ackSeq, ack, queued := t.enqueueReliable(seq, f)
547- if ack {
548- t.queueACK(ackSeq)
549- }
550- if queued {
551- t.signalDelivery()
552- }
553- return
554- }
555- t.deliver(f)
556- }
557-}
558-
559-func (t *Transport) enqueueReliable(seq uint64, f ports.Frame) (ackSeq uint64, ack bool, queued bool) {
560- t.deliverMu.Lock()
561- defer t.deliverMu.Unlock()
562- if seq < t.nextRecvSeq {
563- return t.highestContiguousRecvLocked(), true, false
564- }
565- if _, exists := t.recvBuf[seq]; exists {
566- return t.highestContiguousRecvLocked(), true, false
567- }
568- if len(t.recvBuf) >= t.maxRecvBuffer && seq != t.nextRecvSeq {
569- return 0, false, false
570- }
571- t.recvBuf[seq] = f
572- return t.highestContiguousRecvLocked(), true, true
573-}
574-
575-func (t *Transport) highestContiguousRecvLocked() uint64 {
576- seq := t.nextRecvSeq
577- for {
578- if _, ok := t.recvBuf[seq]; ok {
579- seq++
580- continue
581- }
582- break
583- }
584- return seq - 1
585-}
586-
587-func (t *Transport) signalDelivery() {
588- t.deliverMu.Lock()
589- t.deliverCond.Signal()
590- t.deliverMu.Unlock()
591-}
592-
593-func (t *Transport) deliveryLoop() {
594- for {
595- t.deliverMu.Lock()
596- f, ok := t.recvBuf[t.nextRecvSeq]
597- for !ok {
598- if t.isClosed() {
599- t.deliverMu.Unlock()
600- return
601- }
602- t.deliverCond.Wait()
603- f, ok = t.recvBuf[t.nextRecvSeq]
604- }
605- delete(t.recvBuf, t.nextRecvSeq)
606- t.nextRecvSeq++
607- t.deliverMu.Unlock()
608- t.deliver(f)
609- }
610-}
611-
612-func (t *Transport) isClosed() bool {
613- t.mu.Lock()
614- defer t.mu.Unlock()
615- return t.closed
616-}
617-
618-func (t *Transport) deliver(f ports.Frame) {
619- select {
620- case t.in <- f:
621- case <-t.done:
622- }
623-}
624-
625-func (t *Transport) notifySendWaitersLocked() {
626- if t.sendWake == nil {
627- return
628- }
629- close(t.sendWake)
630- t.sendWake = make(chan struct{})
631-}
632-
633-func (t *Transport) retransmitLoop() {
634- resend := t.clock.NewTimer(t.resendAfter)
635- defer resend.Stop()
636- for {
637- select {
638- case <-resend.C():
639- t.queueRetransmits()
640- resend.Reset(t.resendAfter)
641- t.notifyTimerHook(func() func() { return t.afterRetransmitTimer })
642- case <-t.done:
643- return
644- }
645- }
646-}
647-
648389 // healthLoop owns link-health timing so bulk retransmission cannot delay state
649390 // transitions or socket hopping.
650391 func (t *Transport) healthLoop() {
@@ -662,165 +403,6 @@ func (t *Transport) healthLoop() {
662403 }
663404 }
664405
665-func (t *Transport) dataSendLoop() {
666- pacer := bytePacer{clk: t.clock, afterTimer: func(deadline time.Time) {
667- t.mu.Lock()
668- hook := t.afterDataPaceWait
669- t.mu.Unlock()
670- if hook != nil {
671- hook(deadline)
672- }
673- }}
674- for {
675- select {
676- case job := <-t.dataSend:
677- t.runDataSendJob(&pacer, job, true)
678- case batch := <-t.retransmitWork:
679- t.runRetransmitBatch(&pacer, batch)
680- case <-t.done:
681- return
682- }
683- }
684-}
685-
686-func (t *Transport) runDataSendJob(pacer *bytePacer, job dataSendJob, initial bool) {
687- if initial {
688- t.markPendingSent(job.seq, job.reliable)
689- }
690- _, err := t.writeDataJob(pacer, job, nil)
691- if err != nil {
692- if initial {
693- t.removePending(job.seq, job.reliable)
694- } else {
695- t.markPendingReady(job.seq, job.reliable)
696- }
697- if !errors.Is(err, errPacerClosed) {
698- t.closeWithError(err)
699- }
700- } else {
701- t.markPendingReady(job.seq, job.reliable)
702- }
703- t.mu.Lock()
704- afterDataJob := t.afterDataJob
705- t.mu.Unlock()
706- if afterDataJob != nil {
707- afterDataJob()
708- }
709- if job.done != nil {
710- if errors.Is(err, errPacerClosed) {
711- err = t.closedError()
712- }
713- job.done <- err
714- close(job.done)
715- }
716-}
717-
718-func (t *Transport) writeDataJob(pacer *bytePacer, job dataSendJob, onFirstWrite func()) (bool, error) {
719- if job.reliable && !t.pendingExists(job.seq) {
720- return false, nil
721- }
722- payload := encodeData(job.seq, job.reliable, job.frame)
723- frags, err := pdgram.FragmentPayload(t.nextCounter(), payload, t.mtu-pdgram.HeaderSize-t.codec.Overhead())
724- if err != nil {
725- return false, err
726- }
727- t.mu.Lock()
728- beforeDataPace := t.beforeDataPace
729- t.mu.Unlock()
730- if beforeDataPace != nil {
731- beforeDataPace()
732- }
733- limits := func() (int, int) {
734- t.mu.Lock()
735- defer t.mu.Unlock()
736- return t.congestion.bytesPerSecond(), t.congestion.burstBytes()
737- }
738- emitted := false
739- wireBytes := 0
740- for i, frag := range frags {
741- if job.reliable && !t.pendingExists(job.seq) {
742- return emitted, nil
743- }
744- raw, err := pdgram.MarshalFragment(frag)
745- if err != nil {
746- return emitted, err
747- }
748- pkt := t.codec.Seal(t.sendDir, t.nextCounter(), raw, nil)
749- wireBytes += len(pkt)
750- if job.reliable {
751- t.mu.Lock()
752- if pending := t.pending[job.seq]; pending != nil && pending.wireBytes < wireBytes {
753- pending.wireBytes = wireBytes
754- }
755- t.mu.Unlock()
756- }
757- if err := pacer.wait(t.done, len(pkt), limits); err != nil {
758- return emitted, err
759- }
760- if job.reliable && !t.pendingExists(job.seq) {
761- return emitted, nil
762- }
763- t.mu.Lock()
764- peer := t.peer
765- closed := t.closed
766- closeErr := t.closeErr
767- t.mu.Unlock()
768- if closed {
769- if closeErr != nil {
770- return emitted, closeErr
771- }
772- return emitted, net.ErrClosed
773- }
774- if peer == nil {
775- return emitted, errors.New("dgram: no peer")
776- }
777- if i == len(frags)-1 {
778- t.markPendingFinalWrite(job.seq)
779- }
780- if err := t.writePacket(pkt, peer); err != nil {
781- return emitted, err
782- }
783- if !emitted {
784- emitted = true
785- if onFirstWrite != nil {
786- onFirstWrite()
787- }
788- }
789- }
790- return emitted, nil
791-}
792-
793-func (t *Transport) runRetransmitBatch(pacer *bytePacer, batch []retransmitRecord) {
794- defer func() {
795- t.mu.Lock()
796- hook := t.afterDataJob
797- t.mu.Unlock()
798- if hook != nil {
799- hook()
800- }
801- }()
802- lossApplied := false
803- onFirstWrite := func() {
804- if lossApplied {
805- return
806- }
807- t.mu.Lock()
808- t.congestion.onLoss()
809- t.mu.Unlock()
810- lossApplied = true
811- }
812- for _, r := range batch {
813- job := dataSendJob{seq: r.seq, reliable: true, frame: r.f}
814- _, err := t.writeDataJob(pacer, job, onFirstWrite)
815- if err == nil {
816- t.markPendingReady(r.seq, true)
817- continue
818- }
819- t.markPendingReady(r.seq, true)
820- return
821- }
822-}
823-
824406 func (t *Transport) heartbeatLoop() {
825407 heartbeat := t.clock.NewTimer(t.heartbeat)
826408 defer heartbeat.Stop()
@@ -845,125 +427,6 @@ func (t *Transport) notifyTimerHook(get func() func()) {
845427 }
846428 }
847429
848-func (t *Transport) queueRetransmits() {
849- // Do not mark records in-flight while the one-slot sender queue is full.
850- // There is one producer and the sender only consumes, so capacity cannot be
851- // lost after this check.
852- if len(t.retransmitWork) == cap(t.retransmitWork) {
853- return
854- }
855-
856- now := t.clock.Now()
857- t.mu.Lock()
858- resend, emitDiagnostic := t.selectRetransmitsLocked(now)
859- t.mu.Unlock()
860-
861- if len(resend) > 0 {
862- select {
863- case t.retransmitWork <- resend:
864- case <-t.done:
865- return
866- }
867- }
868- if emitDiagnostic {
869- t.emitDiagnostic()
870- }
871-}
872-
873-// selectRetransmitsLocked deterministically reserves one bounded retransmit
874-// turn. The caller must hold t.mu and performs all pacing and writes after
875-// releasing it.
876-func (t *Transport) selectRetransmitsLocked(now time.Time) ([]retransmitRecord, bool) {
877- resend := make([]retransmitRecord, 0, t.maxResendPerTick)
878- seqs := make([]uint64, 0, len(t.pending))
879- for seq := range t.pending {
880- seqs = append(seqs, seq)
881- }
882- slices.Sort(seqs)
883- limit := t.maxResendPerTick
884- emitDiagnostic := false
885- for _, seq := range seqs {
886- if limit <= 0 {
887- break
888- }
889- p := t.pending[seq]
890- if p == nil || p.last.IsZero() || p.initialInFlight || now.Sub(p.last) < t.resendDelayLocked(p) {
891- continue
892- }
893- p.initialInFlight = true
894- p.attempts++
895- p.retransmitted = true
896- t.retransmits++
897- emitDiagnostic = emitDiagnostic || t.retransmits%64 == 0
898- resend = append(resend, retransmitRecord{seq: seq, f: p.frame})
899- limit--
900- }
901- return resend, emitDiagnostic
902-}
903-
904-// resendPending remains a test hook for retransmission selection. Constructed
905-// transports transfer its batch to the sole data sender; small unit fixtures
906-// without transport goroutines execute the selected writes inline.
907-func (t *Transport) resendPending() {
908- now := t.clock.Now()
909- t.mu.Lock()
910- resend, _ := t.selectRetransmitsLocked(now)
911- t.mu.Unlock()
912- if t.retransmitWork != nil {
913- select {
914- case t.retransmitWork <- resend:
915- case <-t.done:
916- }
917- return
918- }
919- pacer := bytePacer{clk: t.clock}
920- t.runRetransmitBatch(&pacer, resend)
921-}
922-
923-func (t *Transport) resendDelayLocked(p *pending) time.Duration {
924- d := t.rto
925- if d <= 0 {
926- d = t.resendAfter
927- }
928- for i := 0; i < p.attempts && d < t.maxResendAfter; i++ {
929- d *= 2
930- if d > t.maxResendAfter {
931- return t.maxResendAfter
932- }
933- }
934- return d
935-}
936-
937-func (t *Transport) updateRTTLocked(sample time.Duration) {
938- if sample <= 0 {
939- return
940- }
941- if t.srtt <= 0 {
942- t.srtt = sample
943- t.rttvar = sample / 2
944- } else {
945- delta := t.srtt - sample
946- if delta < 0 {
947- delta = -delta
948- }
949- t.rttvar = (3*t.rttvar + delta) / 4
950- t.srtt = (7*t.srtt + sample) / 8
951- }
952- variation := 4 * t.rttvar
953- if variation < time.Millisecond {
954- variation = time.Millisecond
955- }
956- rto := t.srtt + variation
957- if rto < t.resendAfter {
958- rto = t.resendAfter
959- }
960- if rto > t.maxResendAfter {
961- rto = t.maxResendAfter
962- }
963- t.rto = rto
964- t.congestion.onRTT(t.srtt)
965-}
966-
967430 func (t *Transport) hopPacketConnOnce(generation uint64) {
968431 t.mu.Lock()
969432 if t.rebind == nil || t.hoppedOffline || t.closed || t.linkState != ports.LinkStateProbing || t.health.generation != generation {
@@ -1176,27 +639,6 @@ func diagnosticAge(now, then time.Time) time.Duration {
1176639 return now.Sub(then)
1177640 }
1178641
1179-func encodeData(seq uint64, reliable bool, f ports.Frame) []byte {
1180- b := make([]byte, dataRecordHeaderSize+len(f.Payload))
1181- b[0] = recData
1182- binary.BigEndian.PutUint64(b[1:9], seq)
1183- if reliable {
1184- b[9] = 1
1185- }
1186- b[10] = byte(f.Type)
1187- copy(b[dataRecordHeaderSize:], f.Payload)
1188- return b
1189-}
1190-func decodeData(b []byte) (uint64, bool, ports.Frame, bool) {
1191- if len(b) < dataRecordHeaderSize || b[0] != recData {
1192- return 0, false, ports.Frame{}, false
1193- }
1194- if b[11] != 0 {
1195- return 0, false, ports.Frame{}, false
1196- }
1197- return binary.BigEndian.Uint64(b[1:9]), b[9] == 1, ports.Frame{Type: ports.MsgType(b[10]), Payload: append([]byte(nil), b[dataRecordHeaderSize:]...)}, true
1198-}
1199-
1200642 // DatagramTransport marks Transport as a UDP-style datagram transport.
1201643 func (t *Transport) beginRuntimeOperation(start ports.RuntimeMarkKind, bytes uint64) func(bool) {
1202644 if t.runtimeObserver == nil {
@@ -3,10 +3,8 @@ package dgram
3 3
4 import (4 import (
5 "context"5 "context"
6- "encoding/binary"
7 "errors"6 "errors"
8 "net"7 "net"
9- "slices"
10 "sync"8 "sync"
11 "time"9 "time"
12 10
@@ -388,263 +386,6 @@ func (t *Transport) pendingExists(seq uint64) bool {
388 return ok386 return ok
389 }387 }
390 388
391-func (t *Transport) readLoop(pc net.PacketConn) {
392- buf := make([]byte, 64*1024)
393- for {
394- n, addr, err := pc.ReadFrom(buf)
395- if err != nil {
396- t.mu.Lock()
397- current := t.pc == pc
398- closed := t.closed
399- t.mu.Unlock()
400- if current && !closed {
401- t.closeWithError(err)
402- }
403- return
404- }
405- t.recvMu.Lock()
406- counter, pt, err := t.codec.Open(append([]byte(nil), buf[:n]...), t.recvDir, nil, t.replay)
407- if err != nil {
408- t.recvMu.Unlock()
409- continue
410- }
411- frag, err := pdgram.UnmarshalFragment(pt)
412- if err != nil {
413- t.recvMu.Unlock()
414- t.notifyMalformedFragment()
415- continue
416- }
417- payload, complete, err := t.reasm.Add(frag)
418- inflight := t.reasm.Inflight()
419- // Keep the diagnostic mirror ordered with reassembly mutations. Rebind can
420- // overlap old and new read loops, so recvMu must remain held through commit.
421- t.mu.Lock()
422- t.reassemblyInflight = inflight
423- var afterAuthenticated func()
424- if err == nil {
425- now := t.clock.Now()
426- t.health.authenticatedPacket(now)
427- t.lastAuthenticatedPacket = now
428- if !t.peerCounterSet || counter > t.peerCounter {
429- t.peer = addr
430- t.peerCounter = counter
431- t.peerCounterSet = true
432- }
433- afterAuthenticated = t.afterAuthenticatedPacket
434- }
435- t.mu.Unlock()
436- t.recvMu.Unlock()
437- if afterAuthenticated != nil {
438- afterAuthenticated()
439- }
440- if err != nil {
441- t.notifyMalformedFragment()
442- t.notifyPacketProcessed()
443- continue
444- }
445- if !complete {
446- t.notifyPacketProcessed()
447- continue
448- }
449- t.recordCompleteRecord()
450- t.handleRecord(payload)
451- t.checkSilence()
452- t.notifyPacketProcessed()
453- }
454-}
455-
456-func (t *Transport) notifyPacketProcessed() {
457- t.mu.Lock()
458- afterPacketProcessed := t.afterPacketProcessed
459- t.mu.Unlock()
460- if afterPacketProcessed != nil {
461- afterPacketProcessed()
462- }
463-}
464-
465-func (t *Transport) notifyMalformedFragment() {
466- t.mu.Lock()
467- afterMalformed := t.afterMalformedFragment
468- t.mu.Unlock()
469- if afterMalformed != nil {
470- afterMalformed()
471- }
472-}
473-
474-func (t *Transport) recordCompleteRecord() {
475- t.mu.Lock()
476- now := t.clock.Now()
477- t.health.completeRecord(now)
478- t.lastCompleteRecord = now
479- t.mu.Unlock()
480-}
481-
482-func (t *Transport) handleRecord(p []byte) {
483- if len(p) < 1 {
484- return
485- }
486- switch p[0] {
487- case recAck:
488- if len(p) != 9 {
489- return
490- }
491- seq := binary.BigEndian.Uint64(p[1:])
492- t.mu.Lock()
493- acked := false
494- ackedBytes := 0
495- for pendingSeq, p := range t.pending {
496- if pendingSeq > seq {
497- continue
498- }
499- if pendingSeq == seq && p != nil && !p.retransmitted {
500- t.updateRTTLocked(t.clock.Now().Sub(p.first))
501- }
502- if p != nil {
503- ackedBytes += p.wireBytes
504- }
505- delete(t.pending, pendingSeq)
506- acked = true
507- }
508- if acked {
509- t.congestion.onACK(ackedBytes)
510- now := t.clock.Now()
511- t.health.ackProgress(now)
512- if len(t.pending) == 0 {
513- t.health.pendingCleared()
514- }
515- t.lastACKProgress = now
516- t.notifySendWaitersLocked()
517- }
518- t.mu.Unlock()
519- if acked {
520- t.setLinkState(ports.LinkStateConnected, nil)
521- t.emitDiagnostic()
522- }
523- case recProbe:
524- if len(p) != 9 {
525- return
526- }
527- t.queueControl(recPong, binary.BigEndian.Uint64(p[1:]))
528- case recPong:
529- if len(p) != 9 {
530- return
531- }
532- id := binary.BigEndian.Uint64(p[1:])
533- t.mu.Lock()
534- ch := t.probeWait[id]
535- if ch != nil {
536- delete(t.probeWait, id)
537- close(ch)
538- }
539- t.mu.Unlock()
540- case recData:
541- seq, reliable, f, ok := decodeData(p)
542- if !ok {
543- return
544- }
545- if reliable {
546- ackSeq, ack, queued := t.enqueueReliable(seq, f)
547- if ack {
548- t.queueACK(ackSeq)
549- }
550- if queued {
551- t.signalDelivery()
552- }
553- return
554- }
555- t.deliver(f)
556- }
557-}
558-
559-func (t *Transport) enqueueReliable(seq uint64, f ports.Frame) (ackSeq uint64, ack bool, queued bool) {
560- t.deliverMu.Lock()
561- defer t.deliverMu.Unlock()
562- if seq < t.nextRecvSeq {
563- return t.highestContiguousRecvLocked(), true, false
564- }
565- if _, exists := t.recvBuf[seq]; exists {
566- return t.highestContiguousRecvLocked(), true, false
567- }
568- if len(t.recvBuf) >= t.maxRecvBuffer && seq != t.nextRecvSeq {
569- return 0, false, false
570- }
571- t.recvBuf[seq] = f
572- return t.highestContiguousRecvLocked(), true, true
573-}
574-
575-func (t *Transport) highestContiguousRecvLocked() uint64 {
576- seq := t.nextRecvSeq
577- for {
578- if _, ok := t.recvBuf[seq]; ok {
579- seq++
580- continue
581- }
582- break
583- }
584- return seq - 1
585-}
586-
587-func (t *Transport) signalDelivery() {
588- t.deliverMu.Lock()
589- t.deliverCond.Signal()
590- t.deliverMu.Unlock()
591-}
592-
593-func (t *Transport) deliveryLoop() {
594- for {
595- t.deliverMu.Lock()
596- f, ok := t.recvBuf[t.nextRecvSeq]
597- for !ok {
598- if t.isClosed() {
599- t.deliverMu.Unlock()
600- return
601- }
602- t.deliverCond.Wait()
603- f, ok = t.recvBuf[t.nextRecvSeq]
604- }
605- delete(t.recvBuf, t.nextRecvSeq)
606- t.nextRecvSeq++
607- t.deliverMu.Unlock()
608- t.deliver(f)
609- }
610-}
611-
612-func (t *Transport) isClosed() bool {
613- t.mu.Lock()
614- defer t.mu.Unlock()
615- return t.closed
616-}
617-
618-func (t *Transport) deliver(f ports.Frame) {
619- select {
620- case t.in <- f:
621- case <-t.done:
622- }
623-}
624-
625-func (t *Transport) notifySendWaitersLocked() {
626- if t.sendWake == nil {
627- return
628- }
629- close(t.sendWake)
630- t.sendWake = make(chan struct{})
631-}
632-
633-func (t *Transport) retransmitLoop() {
634- resend := t.clock.NewTimer(t.resendAfter)
635- defer resend.Stop()
636- for {
637- select {
638- case <-resend.C():
639- t.queueRetransmits()
640- resend.Reset(t.resendAfter)
641- t.notifyTimerHook(func() func() { return t.afterRetransmitTimer })
642- case <-t.done:
643- return
644- }
645- }
646-}
647-
648 // healthLoop owns link-health timing so bulk retransmission cannot delay state389 // healthLoop owns link-health timing so bulk retransmission cannot delay state
649 // transitions or socket hopping.390 // transitions or socket hopping.
650 func (t *Transport) healthLoop() {391 func (t *Transport) healthLoop() {
@@ -662,165 +403,6 @@ func (t *Transport) healthLoop() {
662 }403 }
663 }404 }
664 405
665-func (t *Transport) dataSendLoop() {
666- pacer := bytePacer{clk: t.clock, afterTimer: func(deadline time.Time) {
667- t.mu.Lock()
668- hook := t.afterDataPaceWait
669- t.mu.Unlock()
670- if hook != nil {
671- hook(deadline)
672- }
673- }}
674- for {
675- select {
676- case job := <-t.dataSend:
677- t.runDataSendJob(&pacer, job, true)
678- case batch := <-t.retransmitWork:
679- t.runRetransmitBatch(&pacer, batch)
680- case <-t.done:
681- return
682- }
683- }
684-}
685-
686-func (t *Transport) runDataSendJob(pacer *bytePacer, job dataSendJob, initial bool) {
687- if initial {
688- t.markPendingSent(job.seq, job.reliable)
689- }
690- _, err := t.writeDataJob(pacer, job, nil)
691- if err != nil {
692- if initial {
693- t.removePending(job.seq, job.reliable)
694- } else {
695- t.markPendingReady(job.seq, job.reliable)
696- }
697- if !errors.Is(err, errPacerClosed) {
698- t.closeWithError(err)
699- }
700- } else {
701- t.markPendingReady(job.seq, job.reliable)
702- }
703- t.mu.Lock()
704- afterDataJob := t.afterDataJob
705- t.mu.Unlock()
706- if afterDataJob != nil {
707- afterDataJob()
708- }
709- if job.done != nil {
710- if errors.Is(err, errPacerClosed) {
711- err = t.closedError()
712- }
713- job.done <- err
714- close(job.done)
715- }
716-}
717-
718-func (t *Transport) writeDataJob(pacer *bytePacer, job dataSendJob, onFirstWrite func()) (bool, error) {
719- if job.reliable && !t.pendingExists(job.seq) {
720- return false, nil
721- }
722- payload := encodeData(job.seq, job.reliable, job.frame)
723- frags, err := pdgram.FragmentPayload(t.nextCounter(), payload, t.mtu-pdgram.HeaderSize-t.codec.Overhead())
724- if err != nil {
725- return false, err
726- }
727- t.mu.Lock()
728- beforeDataPace := t.beforeDataPace
729- t.mu.Unlock()
730- if beforeDataPace != nil {
731- beforeDataPace()
732- }
733- limits := func() (int, int) {
734- t.mu.Lock()
735- defer t.mu.Unlock()
736- return t.congestion.bytesPerSecond(), t.congestion.burstBytes()
737- }
738- emitted := false
739- wireBytes := 0
740- for i, frag := range frags {
741- if job.reliable && !t.pendingExists(job.seq) {
742- return emitted, nil
743- }
744- raw, err := pdgram.MarshalFragment(frag)
745- if err != nil {
746- return emitted, err
747- }
748- pkt := t.codec.Seal(t.sendDir, t.nextCounter(), raw, nil)
749- wireBytes += len(pkt)
750- if job.reliable {
751- t.mu.Lock()
752- if pending := t.pending[job.seq]; pending != nil && pending.wireBytes < wireBytes {
753- pending.wireBytes = wireBytes
754- }
755- t.mu.Unlock()
756- }
757- if err := pacer.wait(t.done, len(pkt), limits); err != nil {
758- return emitted, err
759- }
760- if job.reliable && !t.pendingExists(job.seq) {
761- return emitted, nil
762- }
763- t.mu.Lock()
764- peer := t.peer
765- closed := t.closed
766- closeErr := t.closeErr
767- t.mu.Unlock()
768- if closed {
769- if closeErr != nil {
770- return emitted, closeErr
771- }
772- return emitted, net.ErrClosed
773- }
774- if peer == nil {
775- return emitted, errors.New("dgram: no peer")
776- }
777- if i == len(frags)-1 {
778- t.markPendingFinalWrite(job.seq)
779- }
780- if err := t.writePacket(pkt, peer); err != nil {
781- return emitted, err
782- }
783- if !emitted {
784- emitted = true
785- if onFirstWrite != nil {
786- onFirstWrite()
787- }
788- }
789- }
790- return emitted, nil
791-}
792-
793-func (t *Transport) runRetransmitBatch(pacer *bytePacer, batch []retransmitRecord) {
794- defer func() {
795- t.mu.Lock()
796- hook := t.afterDataJob
797- t.mu.Unlock()
798- if hook != nil {
799- hook()
800- }
801- }()
802- lossApplied := false
803- onFirstWrite := func() {
804- if lossApplied {
805- return
806- }
807- t.mu.Lock()
808- t.congestion.onLoss()
809- t.mu.Unlock()
810- lossApplied = true
811- }
812- for _, r := range batch {
813- job := dataSendJob{seq: r.seq, reliable: true, frame: r.f}
814- _, err := t.writeDataJob(pacer, job, onFirstWrite)
815- if err == nil {
816- t.markPendingReady(r.seq, true)
817- continue
818- }
819- t.markPendingReady(r.seq, true)
820- return
821- }
822-}
823-
824 func (t *Transport) heartbeatLoop() {406 func (t *Transport) heartbeatLoop() {
825 heartbeat := t.clock.NewTimer(t.heartbeat)407 heartbeat := t.clock.NewTimer(t.heartbeat)
826 defer heartbeat.Stop()408 defer heartbeat.Stop()
@@ -845,125 +427,6 @@ func (t *Transport) notifyTimerHook(get func() func()) {
845 }427 }
846 }428 }
847 429
848-func (t *Transport) queueRetransmits() {
849- // Do not mark records in-flight while the one-slot sender queue is full.
850- // There is one producer and the sender only consumes, so capacity cannot be
851- // lost after this check.
852- if len(t.retransmitWork) == cap(t.retransmitWork) {
853- return
854- }
855-
856- now := t.clock.Now()
857- t.mu.Lock()
858- resend, emitDiagnostic := t.selectRetransmitsLocked(now)
859- t.mu.Unlock()
860-
861- if len(resend) > 0 {
862- select {
863- case t.retransmitWork <- resend:
864- case <-t.done:
865- return
866- }
867- }
868- if emitDiagnostic {
869- t.emitDiagnostic()
870- }
871-}
872-
873-// selectRetransmitsLocked deterministically reserves one bounded retransmit
874-// turn. The caller must hold t.mu and performs all pacing and writes after
875-// releasing it.
876-func (t *Transport) selectRetransmitsLocked(now time.Time) ([]retransmitRecord, bool) {
877- resend := make([]retransmitRecord, 0, t.maxResendPerTick)
878- seqs := make([]uint64, 0, len(t.pending))
879- for seq := range t.pending {
880- seqs = append(seqs, seq)
881- }
882- slices.Sort(seqs)
883- limit := t.maxResendPerTick
884- emitDiagnostic := false
885- for _, seq := range seqs {
886- if limit <= 0 {
887- break
888- }
889- p := t.pending[seq]
890- if p == nil || p.last.IsZero() || p.initialInFlight || now.Sub(p.last) < t.resendDelayLocked(p) {
891- continue
892- }
893- p.initialInFlight = true
894- p.attempts++
895- p.retransmitted = true
896- t.retransmits++
897- emitDiagnostic = emitDiagnostic || t.retransmits%64 == 0
898- resend = append(resend, retransmitRecord{seq: seq, f: p.frame})
899- limit--
900- }
901- return resend, emitDiagnostic
902-}
903-
904-// resendPending remains a test hook for retransmission selection. Constructed
905-// transports transfer its batch to the sole data sender; small unit fixtures
906-// without transport goroutines execute the selected writes inline.
907-func (t *Transport) resendPending() {
908- now := t.clock.Now()
909- t.mu.Lock()
910- resend, _ := t.selectRetransmitsLocked(now)
911- t.mu.Unlock()
912- if t.retransmitWork != nil {
913- select {
914- case t.retransmitWork <- resend:
915- case <-t.done:
916- }
917- return
918- }
919- pacer := bytePacer{clk: t.clock}
920- t.runRetransmitBatch(&pacer, resend)
921-}
922-
923-func (t *Transport) resendDelayLocked(p *pending) time.Duration {
924- d := t.rto
925- if d <= 0 {
926- d = t.resendAfter
927- }
928- for i := 0; i < p.attempts && d < t.maxResendAfter; i++ {
929- d *= 2
930- if d > t.maxResendAfter {
931- return t.maxResendAfter
932- }
933- }
934- return d
935-}
936-
937-func (t *Transport) updateRTTLocked(sample time.Duration) {
938- if sample <= 0 {
939- return
940- }
941- if t.srtt <= 0 {
942- t.srtt = sample
943- t.rttvar = sample / 2
944- } else {
945- delta := t.srtt - sample
946- if delta < 0 {
947- delta = -delta
948- }
949- t.rttvar = (3*t.rttvar + delta) / 4
950- t.srtt = (7*t.srtt + sample) / 8
951- }
952- variation := 4 * t.rttvar
953- if variation < time.Millisecond {
954- variation = time.Millisecond
955- }
956- rto := t.srtt + variation
957- if rto < t.resendAfter {
958- rto = t.resendAfter
959- }
960- if rto > t.maxResendAfter {
961- rto = t.maxResendAfter
962- }
963- t.rto = rto
964- t.congestion.onRTT(t.srtt)
965-}
966-
967 func (t *Transport) hopPacketConnOnce(generation uint64) {430 func (t *Transport) hopPacketConnOnce(generation uint64) {
968 t.mu.Lock()431 t.mu.Lock()
969 if t.rebind == nil || t.hoppedOffline || t.closed || t.linkState != ports.LinkStateProbing || t.health.generation != generation {432 if t.rebind == nil || t.hoppedOffline || t.closed || t.linkState != ports.LinkStateProbing || t.health.generation != generation {
@@ -1176,27 +639,6 @@ func diagnosticAge(now, then time.Time) time.Duration {
1176 return now.Sub(then)639 return now.Sub(then)
1177 }640 }
1178 641
1179-func encodeData(seq uint64, reliable bool, f ports.Frame) []byte {
1180- b := make([]byte, dataRecordHeaderSize+len(f.Payload))
1181- b[0] = recData
1182- binary.BigEndian.PutUint64(b[1:9], seq)
1183- if reliable {
1184- b[9] = 1
1185- }
1186- b[10] = byte(f.Type)
1187- copy(b[dataRecordHeaderSize:], f.Payload)
1188- return b
1189-}
1190-func decodeData(b []byte) (uint64, bool, ports.Frame, bool) {
1191- if len(b) < dataRecordHeaderSize || b[0] != recData {
1192- return 0, false, ports.Frame{}, false
1193- }
1194- if b[11] != 0 {
1195- return 0, false, ports.Frame{}, false
1196- }
1197- return binary.BigEndian.Uint64(b[1:9]), b[9] == 1, ports.Frame{Type: ports.MsgType(b[10]), Payload: append([]byte(nil), b[dataRecordHeaderSize:]...)}, true
1198-}
1199-
1200 // DatagramTransport marks Transport as a UDP-style datagram transport.642 // DatagramTransport marks Transport as a UDP-style datagram transport.
1201 func (t *Transport) beginRuntimeOperation(start ports.RuntimeMarkKind, bytes uint64) func(bool) {643 func (t *Transport) beginRuntimeOperation(start ports.RuntimeMarkKind, bytes uint64) func(bool) {
1202 if t.runtimeObserver == nil {644 if t.runtimeObserver == nil {
A internal/adapters/dgram/transport_receive.go +264 -0
@@ -0,0 +1,264 @@
1+package dgram
2+
3+import (
4+ "encoding/binary"
5+ "net"
6+
7+ "github.com/bnema/vev/internal/ports"
8+ pdgram "github.com/bnema/vev/pkg/dgram"
9+)
10+
11+func (t *Transport) readLoop(pc net.PacketConn) {
12+ buf := make([]byte, 64*1024)
13+ for {
14+ n, addr, err := pc.ReadFrom(buf)
15+ if err != nil {
16+ t.mu.Lock()
17+ current := t.pc == pc
18+ closed := t.closed
19+ t.mu.Unlock()
20+ if current && !closed {
21+ t.closeWithError(err)
22+ }
23+ return
24+ }
25+ t.recvMu.Lock()
26+ counter, pt, err := t.codec.Open(append([]byte(nil), buf[:n]...), t.recvDir, nil, t.replay)
27+ if err != nil {
28+ t.recvMu.Unlock()
29+ continue
30+ }
31+ frag, err := pdgram.UnmarshalFragment(pt)
32+ if err != nil {
33+ t.recvMu.Unlock()
34+ t.notifyMalformedFragment()
35+ continue
36+ }
37+ payload, complete, err := t.reasm.Add(frag)
38+ inflight := t.reasm.Inflight()
39+ // Keep the diagnostic mirror ordered with reassembly mutations. Rebind can
40+ // overlap old and new read loops, so recvMu must remain held through commit.
41+ t.mu.Lock()
42+ t.reassemblyInflight = inflight
43+ var afterAuthenticated func()
44+ if err == nil {
45+ now := t.clock.Now()
46+ t.health.authenticatedPacket(now)
47+ t.lastAuthenticatedPacket = now
48+ if !t.peerCounterSet || counter > t.peerCounter {
49+ t.peer = addr
50+ t.peerCounter = counter
51+ t.peerCounterSet = true
52+ }
53+ afterAuthenticated = t.afterAuthenticatedPacket
54+ }
55+ t.mu.Unlock()
56+ t.recvMu.Unlock()
57+ if afterAuthenticated != nil {
58+ afterAuthenticated()
59+ }
60+ if err != nil {
61+ t.notifyMalformedFragment()
62+ t.notifyPacketProcessed()
63+ continue
64+ }
65+ if !complete {
66+ t.notifyPacketProcessed()
67+ continue
68+ }
69+ t.recordCompleteRecord()
70+ t.handleRecord(payload)
71+ t.checkSilence()
72+ t.notifyPacketProcessed()
73+ }
74+}
75+
76+func (t *Transport) notifyPacketProcessed() {
77+ t.mu.Lock()
78+ afterPacketProcessed := t.afterPacketProcessed
79+ t.mu.Unlock()
80+ if afterPacketProcessed != nil {
81+ afterPacketProcessed()
82+ }
83+}
84+
85+func (t *Transport) notifyMalformedFragment() {
86+ t.mu.Lock()
87+ afterMalformed := t.afterMalformedFragment
88+ t.mu.Unlock()
89+ if afterMalformed != nil {
90+ afterMalformed()
91+ }
92+}
93+
94+func (t *Transport) recordCompleteRecord() {
95+ t.mu.Lock()
96+ now := t.clock.Now()
97+ t.health.completeRecord(now)
98+ t.lastCompleteRecord = now
99+ t.mu.Unlock()
100+}
101+
102+func (t *Transport) handleRecord(p []byte) {
103+ if len(p) < 1 {
104+ return
105+ }
106+ switch p[0] {
107+ case recAck:
108+ if len(p) != 9 {
109+ return
110+ }
111+ seq := binary.BigEndian.Uint64(p[1:])
112+ t.mu.Lock()
113+ acked := false
114+ ackedBytes := 0
115+ for pendingSeq, p := range t.pending {
116+ if pendingSeq > seq {
117+ continue
118+ }
119+ if pendingSeq == seq && p != nil && !p.retransmitted {
120+ t.updateRTTLocked(t.clock.Now().Sub(p.first))
121+ }
122+ if p != nil {
123+ ackedBytes += p.wireBytes
124+ }
125+ delete(t.pending, pendingSeq)
126+ acked = true
127+ }
128+ if acked {
129+ t.congestion.onACK(ackedBytes)
130+ now := t.clock.Now()
131+ t.health.ackProgress(now)
132+ if len(t.pending) == 0 {
133+ t.health.pendingCleared()
134+ }
135+ t.lastACKProgress = now
136+ t.notifySendWaitersLocked()
137+ }
138+ t.mu.Unlock()
139+ if acked {
140+ t.setLinkState(ports.LinkStateConnected, nil)
141+ t.emitDiagnostic()
142+ }
143+ case recProbe:
144+ if len(p) != 9 {
145+ return
146+ }
147+ t.queueControl(recPong, binary.BigEndian.Uint64(p[1:]))
148+ case recPong:
149+ if len(p) != 9 {
150+ return
151+ }
152+ id := binary.BigEndian.Uint64(p[1:])
153+ t.mu.Lock()
154+ ch := t.probeWait[id]
155+ if ch != nil {
156+ delete(t.probeWait, id)
157+ close(ch)
158+ }
159+ t.mu.Unlock()
160+ case recData:
161+ seq, reliable, f, ok := decodeData(p)
162+ if !ok {
163+ return
164+ }
165+ if reliable {
166+ ackSeq, ack, queued := t.enqueueReliable(seq, f)
167+ if ack {
168+ t.queueACK(ackSeq)
169+ }
170+ if queued {
171+ t.signalDelivery()
172+ }
173+ return
174+ }
175+ t.deliver(f)
176+ }
177+}
178+
179+func (t *Transport) enqueueReliable(seq uint64, f ports.Frame) (ackSeq uint64, ack bool, queued bool) {
180+ t.deliverMu.Lock()
181+ defer t.deliverMu.Unlock()
182+ if seq < t.nextRecvSeq {
183+ return t.highestContiguousRecvLocked(), true, false
184+ }
185+ if _, exists := t.recvBuf[seq]; exists {
186+ return t.highestContiguousRecvLocked(), true, false
187+ }
188+ if len(t.recvBuf) >= t.maxRecvBuffer && seq != t.nextRecvSeq {
189+ return 0, false, false
190+ }
191+ t.recvBuf[seq] = f
192+ return t.highestContiguousRecvLocked(), true, true
193+}
194+
195+func (t *Transport) highestContiguousRecvLocked() uint64 {
196+ seq := t.nextRecvSeq
197+ for {
198+ if _, ok := t.recvBuf[seq]; ok {
199+ seq++
200+ continue
201+ }
202+ break
203+ }
204+ return seq - 1
205+}
206+
207+func (t *Transport) signalDelivery() {
208+ t.deliverMu.Lock()
209+ t.deliverCond.Signal()
210+ t.deliverMu.Unlock()
211+}
212+
213+func (t *Transport) deliveryLoop() {
214+ for {
215+ t.deliverMu.Lock()
216+ f, ok := t.recvBuf[t.nextRecvSeq]
217+ for !ok {
218+ if t.isClosed() {
219+ t.deliverMu.Unlock()
220+ return
221+ }
222+ t.deliverCond.Wait()
223+ f, ok = t.recvBuf[t.nextRecvSeq]
224+ }
225+ delete(t.recvBuf, t.nextRecvSeq)
226+ t.nextRecvSeq++
227+ t.deliverMu.Unlock()
228+ t.deliver(f)
229+ }
230+}
231+
232+func (t *Transport) isClosed() bool {
233+ t.mu.Lock()
234+ defer t.mu.Unlock()
235+ return t.closed
236+}
237+
238+func (t *Transport) deliver(f ports.Frame) {
239+ select {
240+ case t.in <- f:
241+ case <-t.done:
242+ }
243+}
244+
245+func encodeData(seq uint64, reliable bool, f ports.Frame) []byte {
246+ b := make([]byte, dataRecordHeaderSize+len(f.Payload))
247+ b[0] = recData
248+ binary.BigEndian.PutUint64(b[1:9], seq)
249+ if reliable {
250+ b[9] = 1
251+ }
252+ b[10] = byte(f.Type)
253+ copy(b[dataRecordHeaderSize:], f.Payload)
254+ return b
255+}
256+func decodeData(b []byte) (uint64, bool, ports.Frame, bool) {
257+ if len(b) < dataRecordHeaderSize || b[0] != recData {
258+ return 0, false, ports.Frame{}, false
259+ }
260+ if b[11] != 0 {
261+ return 0, false, ports.Frame{}, false
262+ }
263+ return binary.BigEndian.Uint64(b[1:9]), b[9] == 1, ports.Frame{Type: ports.MsgType(b[10]), Payload: append([]byte(nil), b[dataRecordHeaderSize:]...)}, true
264+}
@@ -0,0 +1,264 @@
1+package dgram
2+
3+import (
4+ "encoding/binary"
5+ "net"
6+
7+ "github.com/bnema/vev/internal/ports"
8+ pdgram "github.com/bnema/vev/pkg/dgram"
9+)
10+
11+func (t *Transport) readLoop(pc net.PacketConn) {
12+ buf := make([]byte, 64*1024)
13+ for {
14+ n, addr, err := pc.ReadFrom(buf)
15+ if err != nil {
16+ t.mu.Lock()
17+ current := t.pc == pc
18+ closed := t.closed
19+ t.mu.Unlock()
20+ if current && !closed {
21+ t.closeWithError(err)
22+ }
23+ return
24+ }
25+ t.recvMu.Lock()
26+ counter, pt, err := t.codec.Open(append([]byte(nil), buf[:n]...), t.recvDir, nil, t.replay)
27+ if err != nil {
28+ t.recvMu.Unlock()
29+ continue
30+ }
31+ frag, err := pdgram.UnmarshalFragment(pt)
32+ if err != nil {
33+ t.recvMu.Unlock()
34+ t.notifyMalformedFragment()
35+ continue
36+ }
37+ payload, complete, err := t.reasm.Add(frag)
38+ inflight := t.reasm.Inflight()
39+ // Keep the diagnostic mirror ordered with reassembly mutations. Rebind can
40+ // overlap old and new read loops, so recvMu must remain held through commit.
41+ t.mu.Lock()
42+ t.reassemblyInflight = inflight
43+ var afterAuthenticated func()
44+ if err == nil {
45+ now := t.clock.Now()
46+ t.health.authenticatedPacket(now)
47+ t.lastAuthenticatedPacket = now
48+ if !t.peerCounterSet || counter > t.peerCounter {
49+ t.peer = addr
50+ t.peerCounter = counter
51+ t.peerCounterSet = true
52+ }
53+ afterAuthenticated = t.afterAuthenticatedPacket
54+ }
55+ t.mu.Unlock()
56+ t.recvMu.Unlock()
57+ if afterAuthenticated != nil {
58+ afterAuthenticated()
59+ }
60+ if err != nil {
61+ t.notifyMalformedFragment()
62+ t.notifyPacketProcessed()
63+ continue
64+ }
65+ if !complete {
66+ t.notifyPacketProcessed()
67+ continue
68+ }
69+ t.recordCompleteRecord()
70+ t.handleRecord(payload)
71+ t.checkSilence()
72+ t.notifyPacketProcessed()
73+ }
74+}
75+
76+func (t *Transport) notifyPacketProcessed() {
77+ t.mu.Lock()
78+ afterPacketProcessed := t.afterPacketProcessed
79+ t.mu.Unlock()
80+ if afterPacketProcessed != nil {
81+ afterPacketProcessed()
82+ }
83+}
84+
85+func (t *Transport) notifyMalformedFragment() {
86+ t.mu.Lock()
87+ afterMalformed := t.afterMalformedFragment
88+ t.mu.Unlock()
89+ if afterMalformed != nil {
90+ afterMalformed()
91+ }
92+}
93+
94+func (t *Transport) recordCompleteRecord() {
95+ t.mu.Lock()
96+ now := t.clock.Now()
97+ t.health.completeRecord(now)
98+ t.lastCompleteRecord = now
99+ t.mu.Unlock()
100+}
101+
102+func (t *Transport) handleRecord(p []byte) {
103+ if len(p) < 1 {
104+ return
105+ }
106+ switch p[0] {
107+ case recAck:
108+ if len(p) != 9 {
109+ return
110+ }
111+ seq := binary.BigEndian.Uint64(p[1:])
112+ t.mu.Lock()
113+ acked := false
114+ ackedBytes := 0
115+ for pendingSeq, p := range t.pending {
116+ if pendingSeq > seq {
117+ continue
118+ }
119+ if pendingSeq == seq && p != nil && !p.retransmitted {
120+ t.updateRTTLocked(t.clock.Now().Sub(p.first))
121+ }
122+ if p != nil {
123+ ackedBytes += p.wireBytes
124+ }
125+ delete(t.pending, pendingSeq)
126+ acked = true
127+ }
128+ if acked {
129+ t.congestion.onACK(ackedBytes)
130+ now := t.clock.Now()
131+ t.health.ackProgress(now)
132+ if len(t.pending) == 0 {
133+ t.health.pendingCleared()
134+ }
135+ t.lastACKProgress = now
136+ t.notifySendWaitersLocked()
137+ }
138+ t.mu.Unlock()
139+ if acked {
140+ t.setLinkState(ports.LinkStateConnected, nil)
141+ t.emitDiagnostic()
142+ }
143+ case recProbe:
144+ if len(p) != 9 {
145+ return
146+ }
147+ t.queueControl(recPong, binary.BigEndian.Uint64(p[1:]))
148+ case recPong:
149+ if len(p) != 9 {
150+ return
151+ }
152+ id := binary.BigEndian.Uint64(p[1:])
153+ t.mu.Lock()
154+ ch := t.probeWait[id]
155+ if ch != nil {
156+ delete(t.probeWait, id)
157+ close(ch)
158+ }
159+ t.mu.Unlock()
160+ case recData:
161+ seq, reliable, f, ok := decodeData(p)
162+ if !ok {
163+ return
164+ }
165+ if reliable {
166+ ackSeq, ack, queued := t.enqueueReliable(seq, f)
167+ if ack {
168+ t.queueACK(ackSeq)
169+ }
170+ if queued {
171+ t.signalDelivery()
172+ }
173+ return
174+ }
175+ t.deliver(f)
176+ }
177+}
178+
179+func (t *Transport) enqueueReliable(seq uint64, f ports.Frame) (ackSeq uint64, ack bool, queued bool) {
180+ t.deliverMu.Lock()
181+ defer t.deliverMu.Unlock()
182+ if seq < t.nextRecvSeq {
183+ return t.highestContiguousRecvLocked(), true, false
184+ }
185+ if _, exists := t.recvBuf[seq]; exists {
186+ return t.highestContiguousRecvLocked(), true, false
187+ }
188+ if len(t.recvBuf) >= t.maxRecvBuffer && seq != t.nextRecvSeq {
189+ return 0, false, false
190+ }
191+ t.recvBuf[seq] = f
192+ return t.highestContiguousRecvLocked(), true, true
193+}
194+
195+func (t *Transport) highestContiguousRecvLocked() uint64 {
196+ seq := t.nextRecvSeq
197+ for {
198+ if _, ok := t.recvBuf[seq]; ok {
199+ seq++
200+ continue
201+ }
202+ break
203+ }
204+ return seq - 1
205+}
206+
207+func (t *Transport) signalDelivery() {
208+ t.deliverMu.Lock()
209+ t.deliverCond.Signal()
210+ t.deliverMu.Unlock()
211+}
212+
213+func (t *Transport) deliveryLoop() {
214+ for {
215+ t.deliverMu.Lock()
216+ f, ok := t.recvBuf[t.nextRecvSeq]
217+ for !ok {
218+ if t.isClosed() {
219+ t.deliverMu.Unlock()
220+ return
221+ }
222+ t.deliverCond.Wait()
223+ f, ok = t.recvBuf[t.nextRecvSeq]
224+ }
225+ delete(t.recvBuf, t.nextRecvSeq)
226+ t.nextRecvSeq++
227+ t.deliverMu.Unlock()
228+ t.deliver(f)
229+ }
230+}
231+
232+func (t *Transport) isClosed() bool {
233+ t.mu.Lock()
234+ defer t.mu.Unlock()
235+ return t.closed
236+}
237+
238+func (t *Transport) deliver(f ports.Frame) {
239+ select {
240+ case t.in <- f:
241+ case <-t.done:
242+ }
243+}
244+
245+func encodeData(seq uint64, reliable bool, f ports.Frame) []byte {
246+ b := make([]byte, dataRecordHeaderSize+len(f.Payload))
247+ b[0] = recData
248+ binary.BigEndian.PutUint64(b[1:9], seq)
249+ if reliable {
250+ b[9] = 1
251+ }
252+ b[10] = byte(f.Type)
253+ copy(b[dataRecordHeaderSize:], f.Payload)
254+ return b
255+}
256+func decodeData(b []byte) (uint64, bool, ports.Frame, bool) {
257+ if len(b) < dataRecordHeaderSize || b[0] != recData {
258+ return 0, false, ports.Frame{}, false
259+ }
260+ if b[11] != 0 {
261+ return 0, false, ports.Frame{}, false
262+ }
263+ return binary.BigEndian.Uint64(b[1:9]), b[9] == 1, ports.Frame{Type: ports.MsgType(b[10]), Payload: append([]byte(nil), b[dataRecordHeaderSize:]...)}, true
264+}
A internal/adapters/dgram/transport_retransmit.go +303 -0
@@ -0,0 +1,303 @@
1+package dgram
2+
3+import (
4+ "errors"
5+ "net"
6+ "slices"
7+ "time"
8+
9+ pdgram "github.com/bnema/vev/pkg/dgram"
10+)
11+
12+func (t *Transport) retransmitLoop() {
13+ resend := t.clock.NewTimer(t.resendAfter)
14+ defer resend.Stop()
15+ for {
16+ select {
17+ case <-resend.C():
18+ t.queueRetransmits()
19+ resend.Reset(t.resendAfter)
20+ t.notifyTimerHook(func() func() { return t.afterRetransmitTimer })
21+ case <-t.done:
22+ return
23+ }
24+ }
25+}
26+
27+func (t *Transport) dataSendLoop() {
28+ pacer := bytePacer{clk: t.clock, afterTimer: func(deadline time.Time) {
29+ t.mu.Lock()
30+ hook := t.afterDataPaceWait
31+ t.mu.Unlock()
32+ if hook != nil {
33+ hook(deadline)
34+ }
35+ }}
36+ for {
37+ select {
38+ case job := <-t.dataSend:
39+ t.runDataSendJob(&pacer, job, true)
40+ case batch := <-t.retransmitWork:
41+ t.runRetransmitBatch(&pacer, batch)
42+ case <-t.done:
43+ return
44+ }
45+ }
46+}
47+
48+func (t *Transport) runDataSendJob(pacer *bytePacer, job dataSendJob, initial bool) {
49+ if initial {
50+ t.markPendingSent(job.seq, job.reliable)
51+ }
52+ _, err := t.writeDataJob(pacer, job, nil)
53+ if err != nil {
54+ if initial {
55+ t.removePending(job.seq, job.reliable)
56+ } else {
57+ t.markPendingReady(job.seq, job.reliable)
58+ }
59+ if !errors.Is(err, errPacerClosed) {
60+ t.closeWithError(err)
61+ }
62+ } else {
63+ t.markPendingReady(job.seq, job.reliable)
64+ }
65+ t.mu.Lock()
66+ afterDataJob := t.afterDataJob
67+ t.mu.Unlock()
68+ if afterDataJob != nil {
69+ afterDataJob()
70+ }
71+ if job.done != nil {
72+ if errors.Is(err, errPacerClosed) {
73+ err = t.closedError()
74+ }
75+ job.done <- err
76+ close(job.done)
77+ }
78+}
79+
80+func (t *Transport) writeDataJob(pacer *bytePacer, job dataSendJob, onFirstWrite func()) (bool, error) {
81+ if job.reliable && !t.pendingExists(job.seq) {
82+ return false, nil
83+ }
84+ payload := encodeData(job.seq, job.reliable, job.frame)
85+ frags, err := pdgram.FragmentPayload(t.nextCounter(), payload, t.mtu-pdgram.HeaderSize-t.codec.Overhead())
86+ if err != nil {
87+ return false, err
88+ }
89+ t.mu.Lock()
90+ beforeDataPace := t.beforeDataPace
91+ t.mu.Unlock()
92+ if beforeDataPace != nil {
93+ beforeDataPace()
94+ }
95+ limits := func() (int, int) {
96+ t.mu.Lock()
97+ defer t.mu.Unlock()
98+ return t.congestion.bytesPerSecond(), t.congestion.burstBytes()
99+ }
100+ emitted := false
101+ wireBytes := 0
102+ for i, frag := range frags {
103+ if job.reliable && !t.pendingExists(job.seq) {
104+ return emitted, nil
105+ }
106+ raw, err := pdgram.MarshalFragment(frag)
107+ if err != nil {
108+ return emitted, err
109+ }
110+ pkt := t.codec.Seal(t.sendDir, t.nextCounter(), raw, nil)
111+ wireBytes += len(pkt)
112+ if job.reliable {
113+ t.mu.Lock()
114+ if pending := t.pending[job.seq]; pending != nil && pending.wireBytes < wireBytes {
115+ pending.wireBytes = wireBytes
116+ }
117+ t.mu.Unlock()
118+ }
119+ if err := pacer.wait(t.done, len(pkt), limits); err != nil {
120+ return emitted, err
121+ }
122+ if job.reliable && !t.pendingExists(job.seq) {
123+ return emitted, nil
124+ }
125+ t.mu.Lock()
126+ peer := t.peer
127+ closed := t.closed
128+ closeErr := t.closeErr
129+ t.mu.Unlock()
130+ if closed {
131+ if closeErr != nil {
132+ return emitted, closeErr
133+ }
134+ return emitted, net.ErrClosed
135+ }
136+ if peer == nil {
137+ return emitted, errors.New("dgram: no peer")
138+ }
139+ if i == len(frags)-1 {
140+ t.markPendingFinalWrite(job.seq)
141+ }
142+ if err := t.writePacket(pkt, peer); err != nil {
143+ return emitted, err
144+ }
145+ if !emitted {
146+ emitted = true
147+ if onFirstWrite != nil {
148+ onFirstWrite()
149+ }
150+ }
151+ }
152+ return emitted, nil
153+}
154+
155+func (t *Transport) runRetransmitBatch(pacer *bytePacer, batch []retransmitRecord) {
156+ defer func() {
157+ t.mu.Lock()
158+ hook := t.afterDataJob
159+ t.mu.Unlock()
160+ if hook != nil {
161+ hook()
162+ }
163+ }()
164+ lossApplied := false
165+ onFirstWrite := func() {
166+ if lossApplied {
167+ return
168+ }
169+ t.mu.Lock()
170+ t.congestion.onLoss()
171+ t.mu.Unlock()
172+ lossApplied = true
173+ }
174+ for _, r := range batch {
175+ job := dataSendJob{seq: r.seq, reliable: true, frame: r.f}
176+ _, err := t.writeDataJob(pacer, job, onFirstWrite)
177+ if err == nil {
178+ t.markPendingReady(r.seq, true)
179+ continue
180+ }
181+ t.markPendingReady(r.seq, true)
182+ return
183+ }
184+}
185+
186+func (t *Transport) queueRetransmits() {
187+ // Do not mark records in-flight while the one-slot sender queue is full.
188+ // There is one producer and the sender only consumes, so capacity cannot be
189+ // lost after this check.
190+ if len(t.retransmitWork) == cap(t.retransmitWork) {
191+ return
192+ }
193+
194+ now := t.clock.Now()
195+ t.mu.Lock()
196+ resend, emitDiagnostic := t.selectRetransmitsLocked(now)
197+ t.mu.Unlock()
198+
199+ if len(resend) > 0 {
200+ select {
201+ case t.retransmitWork <- resend:
202+ case <-t.done:
203+ return
204+ }
205+ }
206+ if emitDiagnostic {
207+ t.emitDiagnostic()
208+ }
209+}
210+
211+// selectRetransmitsLocked deterministically reserves one bounded retransmit
212+// turn. The caller must hold t.mu and performs all pacing and writes after
213+// releasing it.
214+func (t *Transport) selectRetransmitsLocked(now time.Time) ([]retransmitRecord, bool) {
215+ resend := make([]retransmitRecord, 0, t.maxResendPerTick)
216+ seqs := make([]uint64, 0, len(t.pending))
217+ for seq := range t.pending {
218+ seqs = append(seqs, seq)
219+ }
220+ slices.Sort(seqs)
221+ limit := t.maxResendPerTick
222+ emitDiagnostic := false
223+ for _, seq := range seqs {
224+ if limit <= 0 {
225+ break
226+ }
227+ p := t.pending[seq]
228+ if p == nil || p.last.IsZero() || p.initialInFlight || now.Sub(p.last) < t.resendDelayLocked(p) {
229+ continue
230+ }
231+ p.initialInFlight = true
232+ p.attempts++
233+ p.retransmitted = true
234+ t.retransmits++
235+ emitDiagnostic = emitDiagnostic || t.retransmits%64 == 0
236+ resend = append(resend, retransmitRecord{seq: seq, f: p.frame})
237+ limit--
238+ }
239+ return resend, emitDiagnostic
240+}
241+
242+// resendPending remains a test hook for retransmission selection. Constructed
243+// transports transfer its batch to the sole data sender; small unit fixtures
244+// without transport goroutines execute the selected writes inline.
245+func (t *Transport) resendPending() {
246+ now := t.clock.Now()
247+ t.mu.Lock()
248+ resend, _ := t.selectRetransmitsLocked(now)
249+ t.mu.Unlock()
250+ if t.retransmitWork != nil {
251+ select {
252+ case t.retransmitWork <- resend:
253+ case <-t.done:
254+ }
255+ return
256+ }
257+ pacer := bytePacer{clk: t.clock}
258+ t.runRetransmitBatch(&pacer, resend)
259+}
260+
261+func (t *Transport) resendDelayLocked(p *pending) time.Duration {
262+ d := t.rto
263+ if d <= 0 {
264+ d = t.resendAfter
265+ }
266+ for i := 0; i < p.attempts && d < t.maxResendAfter; i++ {
267+ d *= 2
268+ if d > t.maxResendAfter {
269+ return t.maxResendAfter
270+ }
271+ }
272+ return d
273+}
274+
275+func (t *Transport) updateRTTLocked(sample time.Duration) {
276+ if sample <= 0 {
277+ return
278+ }
279+ if t.srtt <= 0 {
280+ t.srtt = sample
281+ t.rttvar = sample / 2
282+ } else {
283+ delta := t.srtt - sample
284+ if delta < 0 {
285+ delta = -delta
286+ }
287+ t.rttvar = (3*t.rttvar + delta) / 4
288+ t.srtt = (7*t.srtt + sample) / 8
289+ }
290+ variation := 4 * t.rttvar
291+ if variation < time.Millisecond {
292+ variation = time.Millisecond
293+ }
294+ rto := t.srtt + variation
295+ if rto < t.resendAfter {
296+ rto = t.resendAfter
297+ }
298+ if rto > t.maxResendAfter {
299+ rto = t.maxResendAfter
300+ }
301+ t.rto = rto
302+ t.congestion.onRTT(t.srtt)
303+}
@@ -0,0 +1,303 @@
1+package dgram
2+
3+import (
4+ "errors"
5+ "net"
6+ "slices"
7+ "time"
8+
9+ pdgram "github.com/bnema/vev/pkg/dgram"
10+)
11+
12+func (t *Transport) retransmitLoop() {
13+ resend := t.clock.NewTimer(t.resendAfter)
14+ defer resend.Stop()
15+ for {
16+ select {
17+ case <-resend.C():
18+ t.queueRetransmits()
19+ resend.Reset(t.resendAfter)
20+ t.notifyTimerHook(func() func() { return t.afterRetransmitTimer })
21+ case <-t.done:
22+ return
23+ }
24+ }
25+}
26+
27+func (t *Transport) dataSendLoop() {
28+ pacer := bytePacer{clk: t.clock, afterTimer: func(deadline time.Time) {
29+ t.mu.Lock()
30+ hook := t.afterDataPaceWait
31+ t.mu.Unlock()
32+ if hook != nil {
33+ hook(deadline)
34+ }
35+ }}
36+ for {
37+ select {
38+ case job := <-t.dataSend:
39+ t.runDataSendJob(&pacer, job, true)
40+ case batch := <-t.retransmitWork:
41+ t.runRetransmitBatch(&pacer, batch)
42+ case <-t.done:
43+ return
44+ }
45+ }
46+}
47+
48+func (t *Transport) runDataSendJob(pacer *bytePacer, job dataSendJob, initial bool) {
49+ if initial {
50+ t.markPendingSent(job.seq, job.reliable)
51+ }
52+ _, err := t.writeDataJob(pacer, job, nil)
53+ if err != nil {
54+ if initial {
55+ t.removePending(job.seq, job.reliable)
56+ } else {
57+ t.markPendingReady(job.seq, job.reliable)
58+ }
59+ if !errors.Is(err, errPacerClosed) {
60+ t.closeWithError(err)
61+ }
62+ } else {
63+ t.markPendingReady(job.seq, job.reliable)
64+ }
65+ t.mu.Lock()
66+ afterDataJob := t.afterDataJob
67+ t.mu.Unlock()
68+ if afterDataJob != nil {
69+ afterDataJob()
70+ }
71+ if job.done != nil {
72+ if errors.Is(err, errPacerClosed) {
73+ err = t.closedError()
74+ }
75+ job.done <- err
76+ close(job.done)
77+ }
78+}
79+
80+func (t *Transport) writeDataJob(pacer *bytePacer, job dataSendJob, onFirstWrite func()) (bool, error) {
81+ if job.reliable && !t.pendingExists(job.seq) {
82+ return false, nil
83+ }
84+ payload := encodeData(job.seq, job.reliable, job.frame)
85+ frags, err := pdgram.FragmentPayload(t.nextCounter(), payload, t.mtu-pdgram.HeaderSize-t.codec.Overhead())
86+ if err != nil {
87+ return false, err
88+ }
89+ t.mu.Lock()
90+ beforeDataPace := t.beforeDataPace
91+ t.mu.Unlock()
92+ if beforeDataPace != nil {
93+ beforeDataPace()
94+ }
95+ limits := func() (int, int) {
96+ t.mu.Lock()
97+ defer t.mu.Unlock()
98+ return t.congestion.bytesPerSecond(), t.congestion.burstBytes()
99+ }
100+ emitted := false
101+ wireBytes := 0
102+ for i, frag := range frags {
103+ if job.reliable && !t.pendingExists(job.seq) {
104+ return emitted, nil
105+ }
106+ raw, err := pdgram.MarshalFragment(frag)
107+ if err != nil {
108+ return emitted, err
109+ }
110+ pkt := t.codec.Seal(t.sendDir, t.nextCounter(), raw, nil)
111+ wireBytes += len(pkt)
112+ if job.reliable {
113+ t.mu.Lock()
114+ if pending := t.pending[job.seq]; pending != nil && pending.wireBytes < wireBytes {
115+ pending.wireBytes = wireBytes
116+ }
117+ t.mu.Unlock()
118+ }
119+ if err := pacer.wait(t.done, len(pkt), limits); err != nil {
120+ return emitted, err
121+ }
122+ if job.reliable && !t.pendingExists(job.seq) {
123+ return emitted, nil
124+ }
125+ t.mu.Lock()
126+ peer := t.peer
127+ closed := t.closed
128+ closeErr := t.closeErr
129+ t.mu.Unlock()
130+ if closed {
131+ if closeErr != nil {
132+ return emitted, closeErr
133+ }
134+ return emitted, net.ErrClosed
135+ }
136+ if peer == nil {
137+ return emitted, errors.New("dgram: no peer")
138+ }
139+ if i == len(frags)-1 {
140+ t.markPendingFinalWrite(job.seq)
141+ }
142+ if err := t.writePacket(pkt, peer); err != nil {
143+ return emitted, err
144+ }
145+ if !emitted {
146+ emitted = true
147+ if onFirstWrite != nil {
148+ onFirstWrite()
149+ }
150+ }
151+ }
152+ return emitted, nil
153+}
154+
155+func (t *Transport) runRetransmitBatch(pacer *bytePacer, batch []retransmitRecord) {
156+ defer func() {
157+ t.mu.Lock()
158+ hook := t.afterDataJob
159+ t.mu.Unlock()
160+ if hook != nil {
161+ hook()
162+ }
163+ }()
164+ lossApplied := false
165+ onFirstWrite := func() {
166+ if lossApplied {
167+ return
168+ }
169+ t.mu.Lock()
170+ t.congestion.onLoss()
171+ t.mu.Unlock()
172+ lossApplied = true
173+ }
174+ for _, r := range batch {
175+ job := dataSendJob{seq: r.seq, reliable: true, frame: r.f}
176+ _, err := t.writeDataJob(pacer, job, onFirstWrite)
177+ if err == nil {
178+ t.markPendingReady(r.seq, true)
179+ continue
180+ }
181+ t.markPendingReady(r.seq, true)
182+ return
183+ }
184+}
185+
186+func (t *Transport) queueRetransmits() {
187+ // Do not mark records in-flight while the one-slot sender queue is full.
188+ // There is one producer and the sender only consumes, so capacity cannot be
189+ // lost after this check.
190+ if len(t.retransmitWork) == cap(t.retransmitWork) {
191+ return
192+ }
193+
194+ now := t.clock.Now()
195+ t.mu.Lock()
196+ resend, emitDiagnostic := t.selectRetransmitsLocked(now)
197+ t.mu.Unlock()
198+
199+ if len(resend) > 0 {
200+ select {
201+ case t.retransmitWork <- resend:
202+ case <-t.done:
203+ return
204+ }
205+ }
206+ if emitDiagnostic {
207+ t.emitDiagnostic()
208+ }
209+}
210+
211+// selectRetransmitsLocked deterministically reserves one bounded retransmit
212+// turn. The caller must hold t.mu and performs all pacing and writes after
213+// releasing it.
214+func (t *Transport) selectRetransmitsLocked(now time.Time) ([]retransmitRecord, bool) {
215+ resend := make([]retransmitRecord, 0, t.maxResendPerTick)
216+ seqs := make([]uint64, 0, len(t.pending))
217+ for seq := range t.pending {
218+ seqs = append(seqs, seq)
219+ }
220+ slices.Sort(seqs)
221+ limit := t.maxResendPerTick
222+ emitDiagnostic := false
223+ for _, seq := range seqs {
224+ if limit <= 0 {
225+ break
226+ }
227+ p := t.pending[seq]
228+ if p == nil || p.last.IsZero() || p.initialInFlight || now.Sub(p.last) < t.resendDelayLocked(p) {
229+ continue
230+ }
231+ p.initialInFlight = true
232+ p.attempts++
233+ p.retransmitted = true
234+ t.retransmits++
235+ emitDiagnostic = emitDiagnostic || t.retransmits%64 == 0
236+ resend = append(resend, retransmitRecord{seq: seq, f: p.frame})
237+ limit--
238+ }
239+ return resend, emitDiagnostic
240+}
241+
242+// resendPending remains a test hook for retransmission selection. Constructed
243+// transports transfer its batch to the sole data sender; small unit fixtures
244+// without transport goroutines execute the selected writes inline.
245+func (t *Transport) resendPending() {
246+ now := t.clock.Now()
247+ t.mu.Lock()
248+ resend, _ := t.selectRetransmitsLocked(now)
249+ t.mu.Unlock()
250+ if t.retransmitWork != nil {
251+ select {
252+ case t.retransmitWork <- resend:
253+ case <-t.done:
254+ }
255+ return
256+ }
257+ pacer := bytePacer{clk: t.clock}
258+ t.runRetransmitBatch(&pacer, resend)
259+}
260+
261+func (t *Transport) resendDelayLocked(p *pending) time.Duration {
262+ d := t.rto
263+ if d <= 0 {
264+ d = t.resendAfter
265+ }
266+ for i := 0; i < p.attempts && d < t.maxResendAfter; i++ {
267+ d *= 2
268+ if d > t.maxResendAfter {
269+ return t.maxResendAfter
270+ }
271+ }
272+ return d
273+}
274+
275+func (t *Transport) updateRTTLocked(sample time.Duration) {
276+ if sample <= 0 {
277+ return
278+ }
279+ if t.srtt <= 0 {
280+ t.srtt = sample
281+ t.rttvar = sample / 2
282+ } else {
283+ delta := t.srtt - sample
284+ if delta < 0 {
285+ delta = -delta
286+ }
287+ t.rttvar = (3*t.rttvar + delta) / 4
288+ t.srtt = (7*t.srtt + sample) / 8
289+ }
290+ variation := 4 * t.rttvar
291+ if variation < time.Millisecond {
292+ variation = time.Millisecond
293+ }
294+ rto := t.srtt + variation
295+ if rto < t.resendAfter {
296+ rto = t.resendAfter
297+ }
298+ if rto > t.maxResendAfter {
299+ rto = t.maxResendAfter
300+ }
301+ t.rto = rto
302+ t.congestion.onRTT(t.srtt)
303+}
M internal/adapters/dgram/transport_send.go +8 -0
@@ -228,3 +228,11 @@ func (t *Transport) writeDatagram(pc net.PacketConn, peer net.Addr, pkt []byte,
228228 }
229229
230230 func (t *Transport) nextCounter() uint64 { t.mu.Lock(); defer t.mu.Unlock(); t.ctr++; return t.ctr }
231+
232+func (t *Transport) notifySendWaitersLocked() {
233+ if t.sendWake == nil {
234+ return
235+ }
236+ close(t.sendWake)
237+ t.sendWake = make(chan struct{})
238+}
@@ -228,3 +228,11 @@ func (t *Transport) writeDatagram(pc net.PacketConn, peer net.Addr, pkt []byte,
228 }228 }
229 229
230 func (t *Transport) nextCounter() uint64 { t.mu.Lock(); defer t.mu.Unlock(); t.ctr++; return t.ctr }230 func (t *Transport) nextCounter() uint64 { t.mu.Lock(); defer t.mu.Unlock(); t.ctr++; return t.ctr }
231+
232+func (t *Transport) notifySendWaitersLocked() {
233+ if t.sendWake == nil {
234+ return
235+ }
236+ close(t.sendWake)
237+ t.sendWake = make(chan struct{})
238+}

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