/
send.go
120 lines (115 loc) · 3.42 KB
/
send.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
package link
import (
"bytes"
"encoding/hex"
"errors"
"fmt"
"io"
"math/rand"
"sync/atomic"
"github.com/fumiama/WireGold/gold/head"
"github.com/fumiama/WireGold/helper"
"github.com/klauspost/compress/zstd"
"github.com/sirupsen/logrus"
)
// WriteAndPut 向 peer 发包并将包放回缓存池
func (l *Link) WriteAndPut(p *head.Packet, istransfer bool) (n int, err error) {
defer p.Put()
teatype := uint8(rand.Intn(16))
sndcnt := atomic.AddUintptr(&l.sendcount, 1)
mtu := l.mtu
if l.mturandomrange > 0 {
mtu -= uint16(rand.Intn(int(l.mturandomrange)))
}
logrus.Debugln("[send] mtu:", mtu, ", count:", sndcnt, ", additional data:", uint16(sndcnt)&0x0fff)
if !istransfer {
l.encrypt(p, uint16(sndcnt), teatype)
}
delta := (int(mtu) - 60) & 0x0000fff8
if delta <= 0 {
logrus.Warnln("[send] reset invalid data frag len", delta, "to 8")
delta = 8
}
if len(p.Data) <= delta {
return l.write(p, teatype, uint16(sndcnt), uint32(len(p.Data)), 0, istransfer, false)
}
if istransfer && p.Flags&0x4000 == 0x4000 && len(p.Data) > delta {
return 0, errors.New("drop don't fragmnet big trans packet")
}
data := p.Data
ttl := p.TTL
totl := uint32(len(data))
pos := 0
packet := head.SelectPacket()
*packet = *p
for ; int(totl)-pos > delta; pos += delta {
logrus.Debugln("[send] split frag [", pos, "~", pos+delta, "], remain:", int(totl)-pos-delta)
packet.Data = data[:delta]
cnt, err := l.write(packet, teatype, uint16(sndcnt), totl, uint16(pos>>3), istransfer, true)
n += cnt
if err != nil {
return n, err
}
data = data[delta:]
packet.TTL = ttl
}
packet.Put()
if len(data) > 0 {
p.Data = data
cnt := 0
cnt, err = l.write(p, teatype, uint16(sndcnt), totl, uint16(pos>>3), istransfer, false)
n += cnt
}
return n, err
}
func (l *Link) encrypt(p *head.Packet, sndcnt uint16, teatype uint8) {
p.FillHash()
logrus.Debugln("[send] data len before encrypt:", len(p.Data))
if l.usezstd {
w := helper.SelectWriter()
defer helper.PutWriter(w)
enc, _ := zstd.NewWriter(w, zstd.WithEncoderLevel(zstd.SpeedFastest))
_, _ = io.Copy(enc, bytes.NewReader(p.Data))
enc.Close()
p.Data = w.Bytes()
logrus.Debugln("[send] data len after zstd:", len(p.Data))
}
if l.aead != nil {
p.Data = l.EncodePreshared(sndcnt&0x0fff, p.Data)
logrus.Debugln("[send] data len after xchacha20:", len(p.Data))
}
p.Data = l.Encode(teatype, p.Data)
logrus.Debugln("[send] data len after tea:", len(p.Data))
}
// write 向 peer 发一个包
func (l *Link) write(p *head.Packet, teatype uint8, additional uint16, datasz uint32, offset uint16, istransfer, hasmore bool) (n int, err error) {
var d []byte
var cl func()
if istransfer {
d, cl = p.Marshal(nil, teatype, additional, 0, 0, false, false)
} else {
d, cl = p.Marshal(l.me.me, teatype, additional, datasz, offset, false, hasmore)
}
if d == nil {
return 0, errors.New("[send] ttl exceeded")
}
if err == nil {
peerep := l.endpoint
if peerep == nil {
return 0, errors.New("[send] nil endpoint of " + p.Dst.String())
}
bound := 64
endl := "..."
if len(d) < bound {
bound = len(d)
endl = "."
}
logrus.Debugln("[send] write", len(d), "bytes data from ep", l.me.myep.LocalAddr(), "to", peerep, "offset:", fmt.Sprintf("%04x", offset))
logrus.Debugln("[send] data bytes", hex.EncodeToString(d[:bound]), endl)
d = l.me.xorenc(d)
logrus.Debugln("[send] data xored", hex.EncodeToString(d[:bound]), endl)
n, err = l.me.myep.WriteToUDP(d, peerep)
cl()
}
return
}