Additional locking to protect msg props actions - #53

Author: matrober-uk

memoryleaks_test.go

@@ -15,18 +15,33 @@ import (
 	"testing"
 	"time"
 
+	"github.com/ibm-messaging/mq-golang-jms20/jms20subset"
 	"github.com/ibm-messaging/mq-golang-jms20/mqjms"
 	"github.com/stretchr/testify/assert"
 )
 
 /*
  * Test for memory leak when there is no message to be received.
+ *
+ * This test is not included in the normal bucket as it sends an enormous number of
+ * messages, and requires human observation of the total process size to establish whether
+ * it passes or not, so can only be run under human supervision
  */
-func DONTRUN_TestLeakOnEmptyGet(t *testing.T) {
+func DONT_RUNTestLeakOnEmptyGet(t *testing.T) {
 
 	// Loads CF parameters from connection_info.json and applicationApiKey.json in the Downloads directory
-	cf, cfErr := mqjms.CreateConnectionFactoryFromDefaultJSONFiles()
-	assert.Nil(t, cfErr)
+	//cf, cfErr := mqjms.CreateConnectionFactoryFromDefaultJSONFiles()
+	//assert.Nil(t, cfErr)
+
+	// Initialise the attributes of the CF in whatever way you like
+	cf := mqjms.ConnectionFactoryImpl{
+		QMName:      "QM1",
+		Hostname:    "localhost",
+		PortNumber:  1414,
+		ChannelName: "DEV.APP.SVRCONN",
+		UserName:    "app",
+		Password:    "passw0rd",
+	}
 
 	// Creates a connection to the queue manager, using defer to close it automatically
 	// at the end of the function (if it was created successfully)
@@ -47,7 +62,6 @@ func DONTRUN_TestLeakOnEmptyGet(t *testing.T) {
 
 	for i := 1; i < 35000; i++ {
 
-		//time.Sleep(100 * time.Millisecond)
 		rcvMsg, errRvc := consumer.ReceiveNoWait()
 		assert.Nil(t, errRvc)
 		assert.Nil(t, rcvMsg)
@@ -64,3 +78,90 @@ func DONTRUN_TestLeakOnEmptyGet(t *testing.T) {
 	time.Sleep(30 * time.Second)
 
 }
+
+/*
+ * Test for memory leak when sending and receiving messages
+ *
+ * This test is not included in the normal bucket as it sends an enormous number of
+ * messages, and requires human observation of the total process size to establish whether
+ * it passes or not, so can only be run under human supervision
+ */
+func DONTRUN_TestLeakOnPutGet(t *testing.T) {
+
+	// Loads CF parameters from connection_info.json and applicationApiKey.json in the Downloads directory
+	//cf, cfErr := mqjms.CreateConnectionFactoryFromDefaultJSONFiles()
+	//assert.Nil(t, cfErr)
+
+	// Initialise the attributes of the CF in whatever way you like
+	cf := mqjms.ConnectionFactoryImpl{
+		QMName:      "QM1",
+		Hostname:    "localhost",
+		PortNumber:  1414,
+		ChannelName: "DEV.APP.SVRCONN",
+		UserName:    "app",
+		Password:    "passw0rd",
+	}
+
+	// Creates a connection to the queue manager, using defer to close it automatically
+	// at the end of the function (if it was created successfully)
+	context, ctxErr := cf.CreateContext()
+	assert.Nil(t, ctxErr)
+	if context != nil {
+		defer context.Close()
+	}
+
+	// Now send the message and get it back again, to check that it roundtripped.
+	queue := context.CreateQueue("DEV.QUEUE.1")
+
+	consumer, errCons := context.CreateConsumer(queue)
+	if consumer != nil {
+		defer consumer.Close()
+	}
+	assert.Nil(t, errCons)
+
+	ttlMillis := 20000
+	producer := context.CreateProducer().SetTimeToLive(ttlMillis)
+
+	for i := 1; i < 25000; i++ {
+
+		// Create a TextMessage and check that we can populate it
+		msgBody := "Message " + fmt.Sprint(i)
+		txtMsg := context.CreateTextMessage()
+		txtMsg.SetText(msgBody)
+		txtMsg.SetIntProperty("MessageNumber", i)
+
+		errSend := producer.Send(queue, txtMsg)
+		assert.Nil(t, errSend)
+
+		rcvMsg, errRvc := consumer.ReceiveNoWait()
+		assert.Nil(t, errRvc)
+		assert.NotNil(t, rcvMsg)
+
+		// Check message body.
+		switch msg := rcvMsg.(type) {
+		case jms20subset.TextMessage:
+			assert.Equal(t, msgBody, *msg.GetText())
+		default:
+			assert.Fail(t, "Got something other than a text message")
+		}
+
+		// Check messageID
+		assert.Equal(t, txtMsg.GetJMSMessageID(), rcvMsg.GetJMSMessageID())
+
+		// Check int property
+		rcvMsgNum, propErr := rcvMsg.GetIntProperty("MessageNumber")
+		assert.Nil(t, propErr)
+		assert.Equal(t, i, rcvMsgNum)
+
+		if i%1000 == 0 {
+			fmt.Println("Messages:", i)
+		}
+
+	}
+
+	fmt.Println("Finished receive calls - waiting for cooldown.")
+	runtime.GC()
+
+	time.Sleep(30 * time.Second)
+
+}

mqjms/ConsumerImpl.go

@@ -129,6 +129,7 @@ func (consumer ConsumerImpl) receiveInternal(gmo *ibmmq.MQGMO) (jms20subset.Mess
 				MessageImpl: MessageImpl{
 					mqmd:      getmqmd,
 					msgHandle: &thisMsgHandle,
+					ctxLock:   consumer.ctx.ctxLock,
 				},
 			}
 
@@ -146,6 +147,7 @@ func (consumer ConsumerImpl) receiveInternal(gmo *ibmmq.MQGMO) (jms20subset.Mess
 				MessageImpl: MessageImpl{
 					mqmd:      getmqmd,
 					msgHandle: &thisMsgHandle,
+					ctxLock:   consumer.ctx.ctxLock,
 				},
 			}
 		}
@@ -404,6 +406,12 @@ func applySelector(selector string, getmqmd *ibmmq.MQMD, gmo *ibmmq.MQGMO) error
 func (consumer ConsumerImpl) Close() {
 
 	if (ibmmq.MQObject{}) != consumer.qObject {
+
+		// Lock the context while we are making calls to the queue manager so that it
+		// doesn't conflict with the finalizer we use (below) to delete unused MessageHandles.
+		consumer.ctx.ctxLock.Lock()
+		defer consumer.ctx.ctxLock.Unlock()
+
 		consumer.qObject.Close(0)
 	}
 

mqjms/ContextImpl.go

@@ -68,7 +68,7 @@ func (ctx ContextImpl) CreateConsumer(dest jms20subset.Destination) (jms20subset
 func (ctx ContextImpl) CreateConsumerWithSelector(dest jms20subset.Destination, selector string) (jms20subset.JMSConsumer, jms20subset.JMSException) {
 
 	// Lock the context while we are making calls to the queue manager so that it
-	// doesn't conflict with the finalizer we use (below) to delete unused MessageHandles.
+	// doesn't conflict with the finalizer we use to delete unused MessageHandles.
 	ctx.ctxLock.Lock()
 	defer ctx.ctxLock.Unlock()
 
@@ -126,7 +126,7 @@ func (ctx ContextImpl) CreateConsumerWithSelector(dest jms20subset.Destination,
 func (ctx ContextImpl) CreateBrowser(dest jms20subset.Destination) (jms20subset.QueueBrowser, jms20subset.JMSException) {
 
 	// Lock the context while we are making calls to the queue manager so that it
-	// doesn't conflict with the finalizer we use (below) to delete unused MessageHandles.
+	// doesn't conflict with the finalizer we use to delete unused MessageHandles.
 	ctx.ctxLock.Lock()
 	defer ctx.ctxLock.Unlock()
 
@@ -177,18 +177,14 @@ func (ctx ContextImpl) CreateBrowser(dest jms20subset.Destination) (jms20subset.
 // CreateTextMessage is a JMS standard mechanism for creating a TextMessage.
 func (ctx ContextImpl) CreateTextMessage() jms20subset.TextMessage {
 
-	// Lock the context while we are making calls to the queue manager so that it
-	// doesn't conflict with the finalizer we use (below) to delete unused MessageHandles.
-	ctx.ctxLock.Lock()
-	defer ctx.ctxLock.Unlock()
-
 	var bodyStr *string
 	thisMsgHandle := ctx.createMsgHandle(ctx.qMgr)
 
 	return &TextMessageImpl{
 		bodyStr: bodyStr,
 		MessageImpl: MessageImpl{
 			msgHandle: &thisMsgHandle,
+			ctxLock:   ctx.ctxLock,
 		},
 	}
 }
@@ -197,6 +193,11 @@ func (ctx ContextImpl) CreateTextMessage() jms20subset.TextMessage {
 // store and retrieve message properties.
 func (ctx ContextImpl) createMsgHandle(qMgr ibmmq.MQQueueManager) ibmmq.MQMessageHandle {
 
+	// Lock the context while we are making calls to the queue manager so that it
+	// doesn't conflict with the finalizer we use to delete unused MessageHandles.
+	ctx.ctxLock.Lock()
+	defer ctx.ctxLock.Unlock()
+
 	cmho := ibmmq.NewMQCMHO()
 	thisMsgHandle, err := qMgr.CrtMH(cmho)
 
@@ -220,17 +221,13 @@ func (ctx ContextImpl) createMsgHandle(qMgr ibmmq.MQQueueManager) ibmmq.MQMessag
 // and initialise it with the chosen text string.
 func (ctx ContextImpl) CreateTextMessageWithString(txt string) jms20subset.TextMessage {
 
-	// Lock the context while we are making calls to the queue manager so that it
-	// doesn't conflict with the finalizer we use (below) to delete unused MessageHandles.
-	ctx.ctxLock.Lock()
-	defer ctx.ctxLock.Unlock()
-
 	thisMsgHandle := ctx.createMsgHandle(ctx.qMgr)
 
 	msg := &TextMessageImpl{
 		bodyStr: &txt,
 		MessageImpl: MessageImpl{
 			msgHandle: &thisMsgHandle,
+			ctxLock:   ctx.ctxLock,
 		},
 	}
 
@@ -240,51 +237,44 @@ func (ctx ContextImpl) CreateTextMessageWithString(txt string) jms20subset.TextM
 // CreateBytesMessage is a JMS standard mechanism for creating a BytesMessage.
 func (ctx ContextImpl) CreateBytesMessage() jms20subset.BytesMessage {
 
-	// Lock the context while we are making calls to the queue manager so that it
-	// doesn't conflict with the finalizer we use (below) to delete unused MessageHandles.
-	ctx.ctxLock.Lock()
-	defer ctx.ctxLock.Unlock()
-
 	var thisBodyBytes *[]byte
 	thisMsgHandle := ctx.createMsgHandle(ctx.qMgr)
 
 	return &BytesMessageImpl{
 		bodyBytes: thisBodyBytes,
 		MessageImpl: MessageImpl{
 			msgHandle: &thisMsgHandle,
+			ctxLock:   ctx.ctxLock,
 		},
 	}
 }
 
 // CreateBytesMessageWithBytes is a JMS standard mechanism for creating a BytesMessage.
 func (ctx ContextImpl) CreateBytesMessageWithBytes(bytes []byte) jms20subset.BytesMessage {
 
-	// Lock the context while we are making calls to the queue manager so that it
-	// doesn't conflict with the finalizer we use (below) to delete unused MessageHandles.
-	ctx.ctxLock.Lock()
-	defer ctx.ctxLock.Unlock()
-
 	thisMsgHandle := ctx.createMsgHandle(ctx.qMgr)
 
 	return &BytesMessageImpl{
 		bodyBytes: &bytes,
 		MessageImpl: MessageImpl{
 			msgHandle: &thisMsgHandle,
+			ctxLock:   ctx.ctxLock,
 		},
 	}
 }
 
 // Commit confirms all messages that were sent under this transaction.
 func (ctx ContextImpl) Commit() jms20subset.JMSException {
 
-	// Lock the context while we are making calls to the queue manager so that it
-	// doesn't conflict with the finalizer we use (below) to delete unused MessageHandles.
-	ctx.ctxLock.Lock()
-	defer ctx.ctxLock.Unlock()
-
 	var retErr jms20subset.JMSException
 
 	if (ibmmq.MQQueueManager{}) != ctx.qMgr {
+
+		// Lock the context while we are making calls to the queue manager so that it
+		// doesn't conflict with the finalizer we use to delete unused MessageHandles.
+		ctx.ctxLock.Lock()
+		defer ctx.ctxLock.Unlock()
+
 		err := ctx.qMgr.Cmit()
 
 		if err != nil {
@@ -336,14 +326,15 @@ func (ctx ContextImpl) Commit() jms20subset.JMSException {
 // Rollback releases all messages that were sent under this transaction.
 func (ctx ContextImpl) Rollback() jms20subset.JMSException {
 
-	// Lock the context while we are making calls to the queue manager so that it
-	// doesn't conflict with the finalizer we use (below) to delete unused MessageHandles.
-	ctx.ctxLock.Lock()
-	defer ctx.ctxLock.Unlock()
-
 	var retErr jms20subset.JMSException
 
 	if (ibmmq.MQQueueManager{}) != ctx.qMgr {
+
+		// Lock the context while we are making calls to the queue manager so that it
+		// doesn't conflict with the finalizer we use to delete unused MessageHandles.
+		ctx.ctxLock.Lock()
+		defer ctx.ctxLock.Unlock()
+
 		err := ctx.qMgr.Back()
 
 		if err != nil {
@@ -370,7 +361,7 @@ func (ctx ContextImpl) Close() {
 	if (ibmmq.MQQueueManager{}) != ctx.qMgr {
 
 		// Lock the context while we are making calls to the queue manager so that it
-		// doesn't conflict with the finalizer we use (below) to delete unused MessageHandles.
+		// doesn't conflict with the finalizer we use to delete unused MessageHandles.
 		ctx.ctxLock.Lock()
 		defer ctx.ctxLock.Unlock()