Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Buffer optimizations #355

Merged
merged 4 commits into from
Jan 10, 2024
Merged

Buffer optimizations #355

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
c3d7299
Avoid creating buffer if pkt larger than ClientNetWriteBufferSize
thedevop Dec 28, 2023
0d10ff8
Use mempool for Properties Encode
thedevop Dec 28, 2023
7d7f98b
Use the more efficient Write instead of Write for Buffer to Buffer write
thedevop Dec 28, 2023
d135409
Merge branch 'main' into buffer_optimization
mochi-co Jan 10, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
15 changes: 9 additions & 6 deletions clients.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -582,7 +582,7 @@ func (cl *Client) WritePacket(pk packets.Packet) error {
return packets.ErrPacketTooLarge // [MQTT-3.1.2-24] [MQTT-3.1.2-25]
}

n, err := func() (n int64, err error) {
n, err := func() (int64, error) {
cl.Lock()
defer cl.Unlock()
if len(cl.State.outbound) == 0 {
Expand All @@ -591,23 +591,26 @@ func (cl *Client) WritePacket(pk packets.Packet) error {
}

// first write to buffer, then flush buffer
n, _ = buf.WriteTo(cl.Net.outbuf) // will always be successful
n, _ := cl.Net.outbuf.Write(buf.Bytes()) // will always be successful
err = cl.flushOutbuf()
return
return int64(n), err
}

// there are more writes in the queue
if cl.Net.outbuf == nil {
if buf.Len() >= cl.ops.options.ClientNetWriteBufferSize {
return buf.WriteTo(cl.Net.Conn)
}
cl.Net.outbuf = new(bytes.Buffer)
}

n, _ = buf.WriteTo(cl.Net.outbuf) // will always be successful
n, _ := cl.Net.outbuf.Write(buf.Bytes()) // will always be successful
if cl.Net.outbuf.Len() < cl.ops.options.ClientNetWriteBufferSize {
return
return int64(n), nil
}

err = cl.flushOutbuf()
return
return int64(n), err
}()
if err != nil {
return err
Expand Down
21 changes: 11 additions & 10 deletions packets/packets.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -348,7 +348,7 @@ func (pk *Packet) ConnectEncode(buf *bytes.Buffer) error {

pk.FixedHeader.Remaining = nb.Len()
pk.FixedHeader.Encode(buf)
_, _ = nb.WriteTo(buf)
buf.Write(nb.Bytes())

return nil
}
Expand Down Expand Up @@ -512,7 +512,8 @@ func (pk *Packet) ConnackEncode(buf *bytes.Buffer) error {

pk.FixedHeader.Remaining = nb.Len()
pk.FixedHeader.Encode(buf)
_, _ = nb.WriteTo(buf)
buf.Write(nb.Bytes())

return nil
}

Expand Down Expand Up @@ -557,7 +558,7 @@ func (pk *Packet) DisconnectEncode(buf *bytes.Buffer) error {

pk.FixedHeader.Remaining = nb.Len()
pk.FixedHeader.Encode(buf)
_, _ = nb.WriteTo(buf)
buf.Write(nb.Bytes())

return nil
}
Expand Down Expand Up @@ -628,7 +629,7 @@ func (pk *Packet) PublishEncode(buf *bytes.Buffer) error {

pk.FixedHeader.Remaining = nb.Len() + len(pk.Payload)
pk.FixedHeader.Encode(buf)
_, _ = nb.WriteTo(buf)
buf.Write(nb.Bytes())
buf.Write(pk.Payload)

return nil
Expand Down Expand Up @@ -719,7 +720,7 @@ func (pk *Packet) encodePubAckRelRecComp(buf *bytes.Buffer) error {

pk.FixedHeader.Remaining = nb.Len()
pk.FixedHeader.Encode(buf)
_, _ = nb.WriteTo(buf)
buf.Write(nb.Bytes())
return nil
}

Expand Down Expand Up @@ -858,7 +859,7 @@ func (pk *Packet) SubackEncode(buf *bytes.Buffer) error {

pk.FixedHeader.Remaining = nb.Len()
pk.FixedHeader.Encode(buf)
_, _ = nb.WriteTo(buf)
buf.Write(nb.Bytes())

return nil
}
Expand Down Expand Up @@ -918,7 +919,7 @@ func (pk *Packet) SubscribeEncode(buf *bytes.Buffer) error {

pk.FixedHeader.Remaining = nb.Len()
pk.FixedHeader.Encode(buf)
_, _ = nb.WriteTo(buf)
buf.Write(nb.Bytes())

return nil
}
Expand Down Expand Up @@ -1014,7 +1015,7 @@ func (pk *Packet) UnsubackEncode(buf *bytes.Buffer) error {

pk.FixedHeader.Remaining = nb.Len()
pk.FixedHeader.Encode(buf)
_, _ = nb.WriteTo(buf)
buf.Write(nb.Bytes())

return nil
}
Expand Down Expand Up @@ -1070,7 +1071,7 @@ func (pk *Packet) UnsubscribeEncode(buf *bytes.Buffer) error {

pk.FixedHeader.Remaining = nb.Len()
pk.FixedHeader.Encode(buf)
_, _ = nb.WriteTo(buf)
buf.Write(nb.Bytes())

return nil
}
Expand Down Expand Up @@ -1132,7 +1133,7 @@ func (pk *Packet) AuthEncode(buf *bytes.Buffer) error {

pk.FixedHeader.Remaining = nb.Len()
pk.FixedHeader.Encode(buf)
_, _ = nb.WriteTo(buf)
buf.Write(nb.Bytes())
return nil
}

Expand Down
12 changes: 8 additions & 4 deletions packets/properties.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,8 @@ import (
"bytes"
"fmt"
"strings"

"github.com/mochi-mqtt/server/v2/mempool"
)

const (
Expand Down Expand Up @@ -199,7 +201,8 @@ func (p *Properties) Encode(pkt byte, mods Mods, b *bytes.Buffer, n int) {
return
}

var buf bytes.Buffer
buf := mempool.GetBuffer()
defer mempool.PutBuffer(buf)
if p.canEncode(pkt, PropPayloadFormat) && p.PayloadFormatFlag {
buf.WriteByte(PropPayloadFormat)
buf.WriteByte(p.PayloadFormat)
Expand Down Expand Up @@ -230,7 +233,7 @@ func (p *Properties) Encode(pkt byte, mods Mods, b *bytes.Buffer, n int) {
for _, v := range p.SubscriptionIdentifier {
if v > 0 {
buf.WriteByte(PropSubscriptionIdentifier)
encodeLength(&buf, int64(v))
encodeLength(buf, int64(v))
}
}
}
Expand Down Expand Up @@ -321,7 +324,8 @@ func (p *Properties) Encode(pkt byte, mods Mods, b *bytes.Buffer, n int) {
}

if !mods.DisallowProblemInfo && p.canEncode(pkt, PropUser) {
pb := bytes.NewBuffer([]byte{})
pb := mempool.GetBuffer()
defer mempool.PutBuffer(pb)
for _, v := range p.User {
pb.WriteByte(PropUser)
pb.Write(encodeString(v.Key))
Expand Down Expand Up @@ -355,7 +359,7 @@ func (p *Properties) Encode(pkt byte, mods Mods, b *bytes.Buffer, n int) {
}

encodeLength(b, int64(buf.Len()))
_, _ = buf.WriteTo(b) // [MQTT-3.1.3-10]
b.Write(buf.Bytes()) // [MQTT-3.1.3-10]
}

// Decode decodes property bytes into a properties struct.
Expand Down