[PATCH] [MAMAC] Update to mamaDispatcher_create to generate unique queue name


Adrienne Ambrose <AAmbrose@...>
 

TESTING -

Attached test app will show memory leak on windows with old version and none in new version.

Running mamalisten with stats enabled will show multiple queues with same name on old version.

After change has been applied each queue will now have a unique name.

 

From 724e766320e96e32576f30208a20ef19ab3ce6bb Mon Sep 17 00:00:00 2001

From: A Ambrose <aambrose@...>

Date: Mon, 12 May 2014 12:35:54 +0100

Subject: [PATCH] Update to mamaDispatcher_create to generate unique queue

name [OMAMA-257]

 

Signed-off-by: A Ambrose <aambrose@...>

---

common/c_cpp/src/c/linux/port.h |  1 +

mama/c_cpp/src/c/queue.c        | 22 ++++++++++++++++++++--

2 files changed, 21 insertions(+), 2 deletions(-)

 

diff --git a/common/c_cpp/src/c/linux/port.h b/common/c_cpp/src/c/linux/port.h

index 79bc7d3..da967b8 100644

--- a/common/c_cpp/src/c/linux/port.h

+++ b/common/c_cpp/src/c/linux/port.h

@@ -129,6 +129,7 @@ int wsem_timedwait (wsem_t* sem, unsigned int ts);

#define wthread_mutex_init      pthread_mutex_init

#define wthread_mutex_unlock    pthread_mutex_unlock

#define wthread_mutex_lock      pthread_mutex_lock

+#define wthread_destroy

#define wthread_mutex_destroy   pthread_mutex_destroy

#define wthread_t               pthread_t

#define wthread_detach          pthread_detach

diff --git a/mama/c_cpp/src/c/queue.c b/mama/c_cpp/src/c/queue.c

index 8fccaa6..92732d0 100644

--- a/mama/c_cpp/src/c/queue.c

+++ b/mama/c_cpp/src/c/queue.c

@@ -115,6 +115,7 @@ typedef struct mamaDisptacherImpl_

     int            mDestroy;

} mamaDispatcherImpl;

+static wInterlockedInt gQueueNumber = 0;

 mama_status

mamaQueue_setClosure ( mamaQueue queue, void* closure)

@@ -192,6 +193,7 @@ mamaQueue_create (mamaQueue* queue,

     mama_status     status      =   MAMA_STATUS_OK;

     mamaBridgeImpl* bImpl       =   (mamaBridgeImpl*)bridgeImpl;

     mamaQueueImpl*  impl        =   NULL;

+    char            queueName[32];

     if (!bridgeImpl)

     {

@@ -205,6 +207,10 @@ mamaQueue_create (mamaQueue* queue,

         return MAMA_STATUS_NULL_ARG;

     }

+    /* Generate a unique queue name */

+    snprintf (queueName, sizeof(queueName),

+            "NO_NAME_%d", wInterlocked_increment(&gQueueNumber));

+

     /*Create the queue structure*/

     impl = (mamaQueueImpl*)calloc (1, sizeof (mamaQueueImpl));

@@ -214,7 +220,7 @@ mamaQueue_create (mamaQueue* queue,

     impl->mBridgeImpl           =   bImpl;

     impl->mMamaQueueBridgeImpl  =   NULL;

     impl->mDispatcher           =   NULL;

-    impl->mQueueName            =   strdup ("NO_NAME");

+    impl->mQueueName            =   strdup (queueName);

     impl->mHighWatermark        =   0;

     impl->mLowWatermark         =   1;

     impl->mQueueMonitorCallbacks.onQueueHighWatermarkExceeded =   NULL;

@@ -398,6 +404,7 @@ mamaQueue_create_usingNative (mamaQueue* queue,

     mama_status     status  =   MAMA_STATUS_OK;

     mamaBridgeImpl* bImpl   =   (mamaBridgeImpl*)bridgeImpl;

     mamaQueueImpl*  impl    =   NULL;

+    char            queueName[32];

     if (!bridgeImpl)

     {

@@ -411,6 +418,10 @@ mamaQueue_create_usingNative (mamaQueue* queue,

         return MAMA_STATUS_NULL_ARG;

     }

+    /* Generate a unique queue name */

+    snprintf (queueName, sizeof(queueName),

+            "NO_NAME_%d", wInterlocked_increment(&gQueueNumber));

+

     /*Create the queue structure*/

     impl = (mamaQueueImpl*)calloc (1, sizeof (mamaQueueImpl));

@@ -421,7 +432,7 @@ mamaQueue_create_usingNative (mamaQueue* queue,

     impl->mBridgeImpl           =   bImpl;

     impl->mMamaQueueBridgeImpl  =   NULL;

     impl->mDispatcher           =   NULL;

-    impl->mQueueName            =   strdup ("NO_NAME");

+    impl->mQueueName            =   strdup (queueName);

     impl->mHighWatermark        =   0;

     impl->mLowWatermark         =   1;

     impl->mQueueMonitorCallbacks.onQueueHighWatermarkExceeded =   NULL;

@@ -748,6 +759,9 @@ mamaQueue_destroy (mamaQueue queue)

         /* Destroy the counter lock */

         wInterlocked_destroy(&impl->mNumberOpenObjects);

+        /* Destroy the queue counter lock */

+        wInterlocked_destroy(&gQueueNumber);

+

         free (impl);

         mama_log (MAMA_LOG_LEVEL_FINEST, "Leaving mamaQueue_destroy for queue 0x%X.", queue);

@@ -1344,7 +1358,11 @@ mamaDispatcher_destroy (mamaDispatcher dispatcher)

     /* Wait for the thread to return. */

     wthread_join (impl->mThread, NULL);

+    /* Destroy the thread handle. */

+    wthread_destroy(impl->mThread);

+

     impl->mQueue->mDispatcher = NULL;

+    impl->mThread = 0;

     free (impl);

     return MAMA_STATUS_OK;

}

--

1.8.3.1

 


This message may contain confidential information and is intended for specific recipients unless explicitly noted otherwise. If you have reason to believe you are not an intended recipient of this message, please delete it and notify the sender. This message may not represent the opinion of IntercontinentalExchange Group, Inc. (ICE), NYSE Euronext or any of their subsidiaries or affiliates, and does not constitute a contract or guarantee. Unencrypted electronic mail is not secure and the recipient of this message is expected to provide safeguards from viruses and pursue alternate means of communication where privacy or a binding message is desired.