CBG-1196 make message.flags an atomic

Author: torcolvin

example/go.mod

@@ -1,6 +1,6 @@
 module github.com/couchbase/go-blip/example
 
-go 1.18
+go 1.23
 
 require (
 	github.com/couchbase/go-blip v0.0.0-00010101000000-000000000000

go.mod

@@ -1,6 +1,6 @@
 module github.com/couchbase/go-blip
 
-go 1.17
+go 1.23
 
 require (
 	github.com/klauspost/compress v1.15.11

message.go

@@ -19,18 +19,19 @@ import (
 	"log"
 	"runtime/debug"
 	"sync"
+	"sync/atomic"
 )
 
 type MessageNumber uint32
 
 // A BLIP message. It could be a request or response or error, and it could be from me or the peer.
 type Message struct {
-	Outgoing   bool          // Is this a message created locally?
-	Sender     *Sender       // The connection that sent this message.
-	Properties Properties    // The message's metadata, similar to HTTP headers.
-	body       []byte        // The message body. MIME type is defined by "Content-Type" property
-	number     MessageNumber // The sequence number of the message in the connection.
-	flags      frameFlags    // Message flags as seen on the first frame.
+	Outgoing   bool                       // Is this a message created locally?
+	Sender     *Sender                    // The connection that sent this message.
+	Properties Properties                 // The message's metadata, similar to HTTP headers.
+	body       []byte                     // The message body. MIME type is defined by "Content-Type" property
+	number     MessageNumber              // The sequence number of the message in the connection.
+	flags      atomic.Pointer[frameFlags] // Message flags as seen on the first frame.
 	bytesSent  uint64
 	bytesAcked uint64
 
@@ -56,7 +57,7 @@ func (message *Message) Close() (err error) {
 
 // Returns a string describing the message for debugging purposes
 func (message *Message) String() string {
-	return frameString(message.number, message.flags)
+	return frameString(message.number, *message.flags.Load())
 }
 
 func frameString(number MessageNumber, flags frameFlags) string {
@@ -72,12 +73,13 @@ func frameString(number MessageNumber, flags frameFlags) string {
 
 // Creates a new outgoing request.
 func NewRequest() *Message {
-	return &Message{
-		flags:      frameFlags(RequestType),
+	m := &Message{
 		Outgoing:   true,
 		Properties: Properties{},
 		cond:       sync.NewCond(&sync.Mutex{}),
 	}
+	m.flags.Store(ptr(frameFlags(RequestType)))
+	return m
 }
 
 // The order in which a request message was sent.
@@ -90,17 +92,17 @@ func (message *Message) SerialNumber() MessageNumber {
 }
 
 // The type of message: request, response or error
-func (message *Message) Type() MessageType { return MessageType(message.flags.messageType()) }
+func (message *Message) Type() MessageType { return MessageType(message.flags.Load().messageType()) }
 
 // True if the message has Urgent priority.
-func (message *Message) Urgent() bool { return message.flags&kUrgent != 0 }
+func (message *Message) Urgent() bool { return *message.flags.Load()&kUrgent != 0 }
 
 // True if the message doesn't want a reply.
-func (message *Message) NoReply() bool { return message.flags&kNoReply != 0 }
+func (message *Message) NoReply() bool { return *message.flags.Load()&kNoReply != 0 }
 
 // True if the message's body was GZIP-compressed in transit.
 // (This is for informative purposes only; you don't need to unzip it yourself!)
-func (message *Message) Compressed() bool { return message.flags&kCompressed != 0 }
+func (message *Message) Compressed() bool { return *message.flags.Load()&kCompressed != 0 }
 
 // Marks an outgoing message as having high priority. Urgent messages get a higher amount of
 // bandwidth. This is useful for streaming media.
@@ -125,11 +127,13 @@ func (request *Message) SetNoReply(noReply bool) {
 
 func (message *Message) setFlag(flag frameFlags, value bool) {
 	message.assertMutable()
+	flags := *message.flags.Load()
 	if value {
-		message.flags |= flag
+		flags |= flag
 	} else {
-		message.flags &^= flag
+		flags &^= flag
 	}
+	message.flags.Store(&flags)
 }
 
 func (message *Message) assertMutable() {
@@ -228,7 +232,7 @@ func (m *Message) SetJSONBodyAsBytes(jsonBytes []byte) {
 // Multiple calls return the same object.
 // If called on a NoReply request, this returns nil.
 func (request *Message) Response() *Message {
-	if request.flags&kNoReply != 0 {
+	if *request.flags.Load()&kNoReply != 0 {
 		return nil
 	}
 	if request.Type() != RequestType {
@@ -257,7 +261,8 @@ func (request *Message) Response() *Message {
 		return request.response
 	}
 	response := request.createResponse()
-	response.flags |= request.flags & kUrgent
+	newFlags := *response.flags.Load() | *request.flags.Load()&kUrgent
+	response.flags.Store(&newFlags)
 	response.Properties = Properties{}
 	request.response = response
 	return response
@@ -271,7 +276,8 @@ func (response *Message) SetError(errDomain string, errCode int, message string)
 		if response.Type() == RequestType {
 			panic("Can't call SetError on a request")
 		}
-		response.flags = (response.flags &^ kTypeMask) | frameFlags(ErrorType)
+		newFlags := *response.flags.Load()&^kTypeMask | frameFlags(ErrorType)
+		response.flags.Store(&newFlags)
 		response.Properties = Properties{
 			"Error-Domain": errDomain,
 			"Error-Code":   fmt.Sprintf("%d", errCode),
@@ -285,13 +291,14 @@ func (response *Message) SetError(errDomain string, errCode int, message string)
 //////// INTERNALS:
 
 func newIncomingMessage(sender *Sender, number MessageNumber, flags frameFlags, reader io.ReadCloser) *Message {
-	return &Message{
+	m := &Message{
 		Sender: sender,
-		flags:  flags | kMoreComing,
 		number: number,
 		reader: reader,
 		cond:   sync.NewCond(&sync.Mutex{}),
 	}
+	m.flags.Store(ptr(flags | kMoreComing))
+	return m
 }
 
 // Creates an incoming message given properties and body; exposed only for testing.
@@ -309,16 +316,17 @@ func NewParsedIncomingMessage(sender *Sender, msgType MessageType, properties Pr
 }
 
 func (request *Message) createResponse() *Message {
+	flags := frameFlags(ResponseType) | (*request.flags.Load() & kUrgent)
 	response := &Message{
-		flags:        frameFlags(ResponseType) | (request.flags & kUrgent),
 		number:       request.number,
 		Outgoing:     !request.Outgoing,
 		inResponseTo: request,
 		cond:         sync.NewCond(&sync.Mutex{}),
 	}
 	if !response.Outgoing {
-		response.flags |= kMoreComing
+		flags |= kMoreComing
 	}
+	response.flags.Store(&flags)
 	return response
 }
 
@@ -409,7 +417,7 @@ func (m *Message) nextFrameToSend(maxSize int) ([]byte, frameFlags) {
 	}
 
 	frame := make([]byte, maxSize)
-	flags := m.flags
+	flags := *m.flags.Load()
 	size, err := io.ReadFull(m.encoder, frame)
 	if err == nil {
 		flags |= kMoreComing
@@ -421,3 +429,7 @@ func (m *Message) nextFrameToSend(maxSize int) ([]byte, frameFlags) {
 
 // A callback function that takes a message and returns nothing
 type MessageCallback func(*Message)
+
+func ptr[T any](v T) *T {
+	return &v
+}

message_test.go

@@ -40,7 +40,7 @@ func TestMessageEncoding(t *testing.T) {
 	assert.Equal(t, "\x25Content-Type\x00ham/rye\x00X-Weather\x00rainy\x00The white knight is sliding down the poker. He balances very badly.", string(serialized))
 	t.Logf("Encoded as %d bytes", len(serialized))
 
-	m2 := newIncomingMessage(nil, 1, m.flags, nil)
+	m2 := newIncomingMessage(nil, 1, *m.flags.Load(), nil)
 	reader := bytes.NewReader(serialized)
 	err = m2.ReadFrom(reader)
 	assert.Equal(t, nil, err)
@@ -63,7 +63,7 @@ func TestMessageEncodingCompressed(t *testing.T) {
 	// goassert.Equals(t, string(serialized), "\x1a\x04\x00ham/rye\x00X-Weather\x00rainy\x00\x1f\x8b\b\x00\x00\tn\x88\x00\xff\f\xca\xd1\t\xc5 \f\x05\xd0U\xee\x04\xce\xf1\x06x\v\xd8z\xd1`\x88ń\x8a\xdb\xd7\xcf\x03\xe7߈\xd5$\x88nR[@\x1c\xaeR\xc4*\xcaX\x868\xe1\x19\x9d3\xe1G\\Y\xb3\xddt\xbc\x9c\xfb\xa8\xe8N_\x00\x00\x00\xff\xffs*\xa1\xa6C\x00\x00\x00")
 	// log.Printf("Encoded compressed as %d bytes", len(serialized))
 
-	m2 := newIncomingMessage(nil, 1, m.flags, nil)
+	m2 := newIncomingMessage(nil, 1, *m.flags.Load(), nil)
 	reader := bytes.NewReader(serialized)
 	err = m2.ReadFrom(reader)
 	assert.Equal(t, nil, err)
@@ -103,7 +103,7 @@ func TestMessageDecoding(t *testing.T) {
 		writer.Close()
 	}()
 
-	incoming := newIncomingMessage(nil, original.number, original.flags, reader)
+	incoming := newIncomingMessage(nil, original.number, *original.flags.Load(), reader)
 	err := incoming.readProperties()
 	assert.Equal(t, nil, err)
 	assert.Equal(t, original.Properties, incoming.Properties)

receiver.go

@@ -291,7 +291,7 @@ func (r *receiver) getPendingResponse(requestNumber MessageNumber, flags frameFl
 	msgStream = r.pendingResponses[requestNumber]
 	if msgStream != nil {
 		if msgStream.bytesWritten == 0 {
-			msgStream.message.flags = flags // set flags based on 1st frame of response
+			msgStream.message.flags.Store(&flags) // set flags based on 1st frame of response
 		}
 		if complete {
 			delete(r.pendingResponses, requestNumber)