saving work with SimpleReader present

Author: Brigitte Lamarche

benchmark_test.go

@@ -231,9 +231,10 @@ func BenchmarkInterpolation(b *testing.B) {
 		},
 		maxAllowedPacket: maxPacketSize,
 		maxWriteSize:     maxPacketSize - 1,
-		buf:              newBuffer(nil),
 	}
-	mc.reader = &mc.buf
+	
+	buf := newBuffer(nil)
+	mc.reader = newSimpleReader(&buf)
 
 	args := []driver.Value{
 		int64(42424242),

buffer.go

@@ -21,7 +21,7 @@ const defaultBufSize = 4096
 // In other words, we can't write and read simultaneously on the same connection.
 // The buffer is similar to bufio.Reader / Writer but zero-copy-ish
 // Also highly optimized for this particular use case.
-type buffer struct {
+type buffer struct { //PROBLEM: figure this all out better
 	buf     []byte
 	nc      net.Conn
 	idx     int
@@ -49,7 +49,7 @@ func (b *buffer) fill(need int) error {
 	// grow buffer if necessary
 	// TODO: let the buffer shrink again at some point
 	//       Maybe keep the org buf slice and swap back?
-	if need > len(b.buf) {
+	if need > len(b.buf) { //look up what len and cap mean again!
 		// Round up to the next multiple of the default size
 		newBuf := make([]byte, ((need/defaultBufSize)+1)*defaultBufSize)
 		copy(newBuf, b.buf)
@@ -92,6 +92,10 @@ func (b *buffer) fill(need int) error {
 // returns next N bytes from buffer.
 // The returned slice is only guaranteed to be valid until the next read
 func (b *buffer) readNext(need int) ([]byte, error) {
+	if need == -1 {
+		return b.takeCompleteBuffer()
+	}
+
 	if b.length < need {
 		// refill
 		if err := b.fill(need); err != nil {
@@ -110,7 +114,7 @@ func (b *buffer) readNext(need int) ([]byte, error) {
 // Otherwise a bigger buffer is made.
 // Only one buffer (total) can be used at a time.
 func (b *buffer) takeBuffer(length int) []byte {
-	if b.length > 0 {
+	if b.length > 0 { //assume its empty
 		return nil
 	}
 
@@ -126,15 +130,17 @@ func (b *buffer) takeBuffer(length int) []byte {
 	return make([]byte, length)
 }
 
+/*
 // shortcut which can be used if the requested buffer is guaranteed to be
 // smaller than defaultBufSize
 // Only one buffer (total) can be used at a time.
 func (b *buffer) takeSmallBuffer(length int) []byte {
-	if b.length == 0 {
+	if b.length == 0 { //assume its empty 
 		return b.buf[:length]
 	}
 	return nil
 }
+*/
 
 // takeCompleteBuffer returns the complete existing buffer.
 // This can be used if the necessary buffer size is unknown.

compress.go

@@ -11,26 +11,37 @@ const (
 )
 
 type compressedReader struct {
-	buf      packetReader
+	buf      *buffer //packetReader
 	bytesBuf []byte
 	mc       *mysqlConn
 	zr       io.ReadCloser
 }
 
+
+type simpleReader struct {
+	buf      *buffer //packetReader
+}
+
 type compressedWriter struct {
 	connWriter io.Writer
 	mc         *mysqlConn
 	zw         *zlib.Writer
 }
 
-func newCompressedReader(buf packetReader, mc *mysqlConn) *compressedReader {
+func newCompressedReader(buf *buffer, mc *mysqlConn) *compressedReader {
 	return &compressedReader{
 		buf:      buf,
 		bytesBuf: make([]byte, 0),
 		mc:       mc,
 	}
 }
 
+func newSimpleReader(buf *buffer) *simpleReader {
+	return &simpleReader{
+		buf:      buf,
+	}
+}
+
 func newCompressedWriter(connWriter io.Writer, mc *mysqlConn) *compressedWriter {
 	return &compressedWriter{
 		connWriter: connWriter,
@@ -52,6 +63,10 @@ func (cr *compressedReader) readNext(need int) ([]byte, error) {
 	return data, nil
 }
 
+func (sr *simpleReader) readNext(need int) ([]byte, error) {
+	return sr.buf.readNext(need)
+}
+
 func (cr *compressedReader) uncompressPacket() error {
 	header, err := cr.buf.readNext(7) // size of compressed header
 

connection.go

@@ -28,7 +28,6 @@ type mysqlContext interface {
 }
 
 type mysqlConn struct {
-	buf                 buffer
 	netConn             net.Conn
 	reader              packetReader
 	writer              io.Writer
@@ -57,6 +56,18 @@ type packetReader interface {
 	readNext(need int) ([]byte, error)
 }
 
+/*
+type packetReadCloser interface{
+	Read(n int) ([]byte, error)
+	Close() error // PROBLEM: is there a way to do this?
+}
+
+type packetWriteCloser interface{
+	Write([]byte) (int, error)
+	Close() error
+}
+*/ 
+
 // Handles parameters set in DSN after the connection is established
 func (mc *mysqlConn) handleParams() (err error) {
 	for param, val := range mc.cfg.Params {
@@ -197,7 +208,11 @@ func (mc *mysqlConn) interpolateParams(query string, args []driver.Value) (strin
 		return "", driver.ErrSkip
 	}
 
-	buf := mc.buf.takeCompleteBuffer()
+	//https://stackoverflow.com/questions/29684609/how-to-check-if-an-object-has-a-particular-method
+
+	//reader has &buf which is a packetreader but also always a buffer
+	buf, _ := mc.reader.readNext(-1) //PROBLEM uncompressed so this works, what if compressed
+
 	if buf == nil {
 		// can not take the buffer. Something must be wrong with the connection
 		errLog.Print(ErrBusyBuffer)

connection_test.go

@@ -15,13 +15,13 @@ import (
 
 func TestInterpolateParams(t *testing.T) {
 	mc := &mysqlConn{
-		buf:              newBuffer(nil),
 		maxAllowedPacket: maxPacketSize,
 		cfg: &Config{
 			InterpolateParams: true,
 		},
 	}
-	mc.reader = &mc.buf
+	buf := newBuffer(nil)
+	mc.reader = newSimpleReader(&buf)
 
 	q, err := mc.interpolateParams("SELECT ?+?", []driver.Value{int64(42), "gopher"})
 	if err != nil {
@@ -36,13 +36,13 @@ func TestInterpolateParams(t *testing.T) {
 
 func TestInterpolateParamsTooManyPlaceholders(t *testing.T) {
 	mc := &mysqlConn{
-		buf:              newBuffer(nil),
 		maxAllowedPacket: maxPacketSize,
 		cfg: &Config{
 			InterpolateParams: true,
 		},
 	}
-	mc.reader = &mc.buf
+	buf := newBuffer(nil)
+	mc.reader = newSimpleReader(&buf)
 
 	q, err := mc.interpolateParams("SELECT ?+?", []driver.Value{int64(42)})
 	if err != driver.ErrSkip {
@@ -54,14 +54,14 @@ func TestInterpolateParamsTooManyPlaceholders(t *testing.T) {
 // https://github.com/go-sql-driver/mysql/pull/490
 func TestInterpolateParamsPlaceholderInString(t *testing.T) {
 	mc := &mysqlConn{
-		buf:              newBuffer(nil),
 		maxAllowedPacket: maxPacketSize,
 		cfg: &Config{
 			InterpolateParams: true,
 		},
 	}
 
-	mc.reader = &mc.buf
+	buf := newBuffer(nil)
+	mc.reader = newSimpleReader(&buf)
 
 	q, err := mc.interpolateParams("SELECT 'abc?xyz',?", []driver.Value{int64(42)})
 	// When InterpolateParams support string literal, this should return `"SELECT 'abc?xyz', 42`

driver.go

@@ -91,15 +91,16 @@ func (d MySQLDriver) Open(dsn string) (driver.Conn, error) {
 		s.startWatcher()
 	}
 
-	mc.buf = newBuffer(mc.netConn)
+	buf := newBuffer(mc.netConn)
+
+	// Set I/O timeouts
+	buf.timeout = mc.cfg.ReadTimeout
+	mc.writeTimeout = mc.cfg.WriteTimeout
 
 	// packet reader and writer in handshake are never compressed
-	mc.reader = &mc.buf
+	mc.reader = newSimpleReader(&buf)
 	mc.writer = mc.netConn
 
-	// Set I/O timeouts
-	mc.buf.timeout = mc.cfg.ReadTimeout
-	mc.writeTimeout = mc.cfg.WriteTimeout
 
 	// Reading Handshake Initialization Packet
 	cipher, err := mc.readInitPacket()
@@ -124,7 +125,7 @@ func (d MySQLDriver) Open(dsn string) (driver.Conn, error) {
 	}
 
 	if mc.cfg.Compress {
-		mc.reader = newCompressedReader(&mc.buf, mc)
+		mc.reader = newCompressedReader(&buf, mc)
 		mc.writer = newCompressedWriter(mc.writer, mc)
 	}
 

packets.go

@@ -28,7 +28,7 @@ func (mc *mysqlConn) readPacket() ([]byte, error) {
 	var prevData []byte
 	for {
 		// read packet header
-		data, err := mc.reader.readNext(4)
+		data, err := mc.reader.readNext(4) 
 		if err != nil {
 			if cerr := mc.canceled.Value(); cerr != nil {
 				return nil, cerr
@@ -64,7 +64,7 @@ func (mc *mysqlConn) readPacket() ([]byte, error) {
 		}
 
 		// read packet body [pktLen bytes]
-		data, err = mc.reader.readNext(pktLen)
+		data, err = mc.reader.readNext(pktLen) 
 		if err != nil {
 			if cerr := mc.canceled.Value(); cerr != nil {
 				return nil, cerr
@@ -283,7 +283,8 @@ func (mc *mysqlConn) writeAuthPacket(cipher []byte) error {
 	}
 
 	// Calculate packet length and get buffer with that size
-	data := mc.buf.takeSmallBuffer(pktLen + 4)
+	data, _ := mc.reader.readNext(pktLen + 4)
+	
 	if data == nil {
 		// can not take the buffer. Something must be wrong with the connection
 		errLog.Print(ErrBusyBuffer)
@@ -326,8 +327,12 @@ func (mc *mysqlConn) writeAuthPacket(cipher []byte) error {
 			return err
 		}
 		mc.netConn = tlsConn
-		mc.buf.nc = tlsConn
+		nc := tlsConn
 
+		// make newBuffer with tls conn, clean slate bc handshake 
+		newBuf := newBuffer(nc)
+		mc.reader = newSimpleReader(&newBuf)
+		
 		mc.writer = mc.netConn
 	}
 
@@ -373,7 +378,8 @@ func (mc *mysqlConn) writeOldAuthPacket(cipher []byte) error {
 
 	// Calculate the packet length and add a tailing 0
 	pktLen := len(scrambleBuff) + 1
-	data := mc.buf.takeSmallBuffer(4 + pktLen)
+	data, _ := mc.reader.readNext(4 + pktLen) 
+
 	if data == nil {
 		// can not take the buffer. Something must be wrong with the connection
 		errLog.Print(ErrBusyBuffer)
@@ -392,7 +398,8 @@ func (mc *mysqlConn) writeOldAuthPacket(cipher []byte) error {
 func (mc *mysqlConn) writeClearAuthPacket() error {
 	// Calculate the packet length and add a tailing 0
 	pktLen := len(mc.cfg.Passwd) + 1
-	data := mc.buf.takeSmallBuffer(4 + pktLen)
+	data, _ := mc.reader.readNext(4 + pktLen)
+	
 	if data == nil {
 		// can not take the buffer. Something must be wrong with the connection
 		errLog.Print(ErrBusyBuffer)
@@ -415,7 +422,8 @@ func (mc *mysqlConn) writeNativeAuthPacket(cipher []byte) error {
 
 	// Calculate the packet length and add a tailing 0
 	pktLen := len(scrambleBuff)
-	data := mc.buf.takeSmallBuffer(4 + pktLen)
+	data, _ := mc.reader.readNext(4 + pktLen)
+	
 	if data == nil {
 		// can not take the buffer. Something must be wrong with the connection
 		errLog.Print(ErrBusyBuffer)
@@ -437,7 +445,8 @@ func (mc *mysqlConn) writeCommandPacket(command byte) error {
 	mc.sequence = 0
 	mc.compressionSequence = 0
 
-	data := mc.buf.takeSmallBuffer(4 + 1)
+	data, _ := mc.reader.readNext(4+1)
+	
 	if data == nil {
 		// can not take the buffer. Something must be wrong with the connection
 		errLog.Print(ErrBusyBuffer)
@@ -457,7 +466,8 @@ func (mc *mysqlConn) writeCommandPacketStr(command byte, arg string) error {
 	mc.compressionSequence = 0
 
 	pktLen := 1 + len(arg)
-	data := mc.buf.takeBuffer(pktLen + 4)
+	data, _ := mc.reader.readNext(pktLen + 4)
+	
 	if data == nil {
 		// can not take the buffer. Something must be wrong with the connection
 		errLog.Print(ErrBusyBuffer)
@@ -479,7 +489,8 @@ func (mc *mysqlConn) writeCommandPacketUint32(command byte, arg uint32) error {
 	mc.sequence = 0
 	mc.compressionSequence = 0
 
-	data := mc.buf.takeSmallBuffer(4 + 1 + 4)
+	data, _ := mc.reader.readNext(4 + 1 + 4)
+	
 	if data == nil {
 		// can not take the buffer. Something must be wrong with the connection
 		errLog.Print(ErrBusyBuffer)
@@ -946,9 +957,9 @@ func (stmt *mysqlStmt) writeExecutePacket(args []driver.Value) error {
 	var data []byte
 
 	if len(args) == 0 {
-		data = mc.buf.takeBuffer(minPktLen)
+		data, _ = mc.reader.readNext(minPktLen)
 	} else {
-		data = mc.buf.takeCompleteBuffer()
+		data, _ = mc.reader.readNext(-1) //how does this work out with compressed?
 	}
 	if data == nil {
 		// can not take the buffer. Something must be wrong with the connection
@@ -1127,7 +1138,8 @@ func (stmt *mysqlStmt) writeExecutePacket(args []driver.Value) error {
 		// In that case we must build the data packet with the new values buffer
 		if valuesCap != cap(paramValues) {
 			data = append(data[:pos], paramValues...)
-			mc.buf.buf = data
+			readerBuffer := mc.reader.getBuffer() //PROBLEM: what the fuck is going on here??
+			readerBuffer.buf = data
 		}
 
 		pos += len(paramValues)