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

MongoDB 3.6: implement the new wire protocol #61

Merged
merged 9 commits into from
Dec 12, 2017
Merged

MongoDB 3.6: implement the new wire protocol #61

Changes from 1 commit
Commits
Show all changes
9 commits
Select commit Hold shift + click to select a range
c1fc70d
test against 3.6
feliixx Nov 10, 2017
0eaf023
update go and mongodb version
feliixx Nov 13, 2017
b5082c2
test fixes on 3.2.17
feliixx Nov 13, 2017
f2ad0ee
fix X509 test
feliixx Nov 14, 2017
3b9a65f
Fix auth test on 3.6-rc3
feliixx Nov 14, 2017
d2ccced
implement OP_MSG wire protocole
feliixx Nov 14, 2017
5e766ed
re-enable TestViewWithCollation
feliixx Dec 4, 2017
145c587
refactor memory pooling
feliixx Dec 6, 2017
019a38f
update to 3.6.0 stable
feliixx Dec 6, 2017
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
36 changes: 12 additions & 24 deletions socket.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -459,10 +459,8 @@ var bytesBufferPool = sync.Pool{

func (socket *mongoSocket) Query(ops ...interface{}) (err error) {

buf := bytesBufferPool.Get().([]byte)
defer func() {
bytesBufferPool.Put(buf[:0])
}()
buf := getSizedBuffer(0)
defer bytesBufferPool.Put(buf)

// Serialize operations synchronously to avoid interrupting
// other goroutines while we can't really be sending data.
Expand Down Expand Up @@ -619,7 +617,8 @@ func (socket *mongoSocket) sendMessage(op *msgOp) (writeCmdResult, error) {
var wr writeCmdResult
var err error

buf := bytesBufferPool.Get().([]byte)
buf := getSizedBuffer(0)
defer bytesBufferPool.Put(buf)

buf = addHeader(buf, dbMessage)
buf = addInt32(buf, int32(op.flags))
Expand Down Expand Up @@ -678,13 +677,6 @@ func (socket *mongoSocket) sendMessage(op *msgOp) (writeCmdResult, error) {
socket.updateDeadline(writeDeadline)
_, err = socket.conn.Write(buf)

// TODO optimize this
// buf[0:8] are always rewritten with setInt32,
// buf[8:12] never change (only used in db response)
// might be interesting to only set byte buf[12] and buf[13]
// to avoid some useless 'append' operations
bytesBufferPool.Put(buf[:0])

if expectReply {
socket.updateDeadline(readDeadline)
wait.Lock()
Expand All @@ -706,15 +698,9 @@ func (socket *mongoSocket) sendMessage(op *msgOp) (writeCmdResult, error) {
return wcr, err
}

var responsePool = sync.Pool{
New: func() interface{} {
return make([]byte, 20)
},
}

// get a slice of byte of `size` length
func getEmptyBody(size int) []byte {
b := responsePool.Get().([]byte)
// get a slice of byte of `size` length from the pool
func getSizedBuffer(size int) []byte {
Copy link

Choose a reason for hiding this comment

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

This is a great idea - makes buffer pooling nice and clean!

b := bytesBufferPool.Get().([]byte)
if len(b) < size {
for i := len(b); i < size; i++ {
b = append(b, byte(0))
Expand Down Expand Up @@ -781,7 +767,9 @@ func (socket *mongoSocket) readLoop() {
socket.kill(err, true)
return
}
b := make([]byte, int(getInt32(s, 0)))
b := getSizedBuffer(int(getInt32(s, 0)))
defer bytesBufferPool.Put(b)

copy(b[0:4], s)

_, err = io.ReadFull(r, b[4:])
Expand Down Expand Up @@ -817,7 +805,8 @@ func (socket *mongoSocket) readLoop() {
socket.Unlock()

case dbMessage:
body := getEmptyBody(int(totalLen) - 16)
body := getSizedBuffer(int(totalLen) - 16)
defer bytesBufferPool.Put(body)
_, err := io.ReadFull(r, body)
if err != nil {
socket.kill(err, true)
Expand Down Expand Up @@ -850,7 +839,6 @@ func (socket *mongoSocket) readLoop() {
socket.kill(fmt.Errorf("couldn't handle response properly"), true)
return
}
responsePool.Put(body)
socket.conn.SetReadDeadline(time.Time{})
default:
socket.kill(errors.New("opcode != 1 && opcode != 2013, corrupted data?"), true)
Expand Down