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

Jump to bottom

[CHANGED] avoid unnecessary data copy in processMsg #788

Merged
merged 2 commits into from Aug 9, 2021
Merged

[CHANGED] avoid unnecessary data copy in processMsg #788

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
f8662df
avoid unnecessary data copy in processMsg
ninjken Aug 5, 2021
9b7c017
reset msgCopied flag in parseState after processMsg() for consistency
ninjken Aug 6, 2021
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Jump to
Jump to file
Failed to load files.
Diff view
Diff view
7 changes: 5 additions & 2 deletions nats.go
View file
Open in desktop
Expand Up @@ -2629,8 +2629,11 @@ func (nc *Conn) processMsg(data []byte) {
// It's possible that we end-up not using the message, but that's ok.

// FIXME(dlc): Need to copy, should/can do COW?
msgPayload := make([]byte, len(data))
copy(msgPayload, data)
var msgPayload = data
if !nc.ps.msgCopied {
msgPayload = make([]byte, len(data))
copy(msgPayload, data)
}

// Check if we have headers encoded here.
var h Header
Expand Down
20 changes: 11 additions & 9 deletions parser.go
View file
Open in desktop
Expand Up @@ -28,14 +28,15 @@ type msgArg struct {
const MAX_CONTROL_LINE_SIZE = 4096

type parseState struct {
state int
as int
drop int
hdr int
ma msgArg
argBuf []byte
msgBuf []byte
scratch [MAX_CONTROL_LINE_SIZE]byte
state int
as int
drop int
hdr int
ma msgArg
argBuf []byte
msgBuf []byte
msgCopied bool
scratch [MAX_CONTROL_LINE_SIZE]byte
}

const (
Expand Down Expand Up @@ -167,7 +168,7 @@ func (nc *Conn) parse(buf []byte) error {
if nc.ps.msgBuf != nil {
if len(nc.ps.msgBuf) >= nc.ps.ma.size {
nc.processMsg(nc.ps.msgBuf)
nc.ps.argBuf, nc.ps.msgBuf, nc.ps.state = nil, nil, MSG_END
nc.ps.argBuf, nc.ps.msgBuf, nc.ps.msgCopied, nc.ps.state = nil, nil, false, MSG_END
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

For consistency, I wonder if we should also reset this boolean here: https://github.com/nats-io/nats.go/pull/788/files#diff-83eb8e32639d01cf443d6d8bde24c1c8be78766090d8c5f8586c36250cfedca6R194
I agree that at this location it means that the message fit in the current read buffer, so it is clearly not copied, so this boolean should not be true..

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

At first I thought it was not needed, but I think I agree with you we should reset the flag as well for code consistency.
PR updated :-)

} else {
// copy as much as we can to the buffer and skip ahead.
toCopy := nc.ps.ma.size - len(nc.ps.msgBuf)
Expand Down Expand Up @@ -403,6 +404,7 @@ func (nc *Conn) parse(buf []byte) error {

nc.ps.msgBuf = make([]byte, lrem, nc.ps.ma.size)
copy(nc.ps.msgBuf, buf[nc.ps.as:])
nc.ps.msgCopied = true
} else {
nc.ps.msgBuf = nc.ps.scratch[len(nc.ps.argBuf):len(nc.ps.argBuf)]
nc.ps.msgBuf = append(nc.ps.msgBuf, (buf[nc.ps.as:])...)
Expand Down