Showing posts with label Clojure. Show all posts
Showing posts with label Clojure. Show all posts

Monday, March 25, 2013

Compare Tibco EMS Performance Under Load with Clojure

We have migrated our Tibco EMS infrastructure from the older Sun V490 servers to the newer HP ProLiant DL380 G7 servers. One of the concerns was how does the new servers perform under load compared with the older servers. I needed to create a load test that compared the performance of the two different types of servers. I turned to Clojure to help me put together this load test and here is the load test code:


Load Test Script in Clojure

; Leveraging Apache Commons to generate the random message
(def msg64 (RandomStringUtils/randomAlphanumeric 64))
(def msg128 (RandomStringUtils/randomAlphanumeric 128))
(def msg256 (RandomStringUtils/randomAlphanumeric 256))
(def msg64k (RandomStringUtils/randomAlphanumeric (* 64 1024)))
(def msg128k (RandomStringUtils/randomAlphanumeric (* 128 1024)))
(def msg256k (RandomStringUtils/randomAlphanumeric (* 256 1024)))
(def msg512k (RandomStringUtils/randomAlphanumeric (* 512 1024)))
(def msg1mb (RandomStringUtils/randomAlphanumeric (* 1024 1024)))
(def persistent-delivery (.intValue (javax.jms.DeliveryMode/PERSISTENT)))


; Send messages
(defn send-messages [server-url user password queue-name data iterations]
  (with-open [connection (-> (TibjmsQueueConnectionFactory. server-url)
            (.createQueueConnection user password))]
    (let [session (.createQueueSession connection false Session/AUTO_ACKNOWLEDGE) 
         queue   (.createQueue session  queue-name)]
      (with-open [sender (.createSender session queue)]
  (dotimes [_ iterations]
          (let [message (.createTextMessage session)]
            (.setJMSDeliveryMode message persistent-delivery)
            (.setText message data)
            (.send sender message)))))))
   

(def test-scenarios (list
 {:label "64 bytes" :msg msg64 :n 5000 }
 {:label "128 bytes" :msg msg128 :n 5000 }
 {:label "256 bytes" :msg msg256 :n 5000 }
 {:label "64KB bytes" :msg msg64k :n 2000 }
 {:label "128KB bytes" :msg msg128k :n 1000 }
 {:label "256KB bytes" :msg msg256k :n 1000 }
 {:label "512KB bytes" :msg msg512k :n 1000 }
 {:label "1MB bytes" :msg msg1mb :n 1000 }))

(doseq [my-test test-scenarios]
  (let [label (:label my-test)
       message (:msg my-test)
       n (:n my-test)]
    (println "\n\n")
 (println (str "Testing for " label " sized messages with n = " n))
 (time (send-messages server-url username password test-queue message n))))

Here's the message consumer part of the load test:


Message Consumer in Clojure

                
; Consume Queue Text messages asynchronously
(defn get-queue-text-messages [server-url user password queue-name process-message]
    (future
        (with-open [connection (-> (TibjmsQueueConnectionFactory. server-url)
                                   (.createQueueConnection user password))]
            (let [session (.createQueueSession connection false Session/AUTO_ACKNOWLEDGE)
                  queue (.createQueue session  queue-name)]
                (with-open [receiver (.createReceiver session queue)]             
                    (.start connection)
                    (loop [acc 0]                      
                        (process-message (.receive receiver) acc)
                        (recur (+ acc 1))))))))

; Dump just the message id      
(defn dump-message-id [message n]
    (println (str n)))

     
; Create function aliases with connection information embedded                   
(defn consume-messages [queue-name message-processor]
    (get-queue-text-messages  server-url username password queue-name message-processor))
 
; Start consuming messages asynchronously
(consume-messages queueName dump-message-id)     

Here are the test results:


Load Test Result on Sun V490

Message SizeNumber of MessagesProcess Time (ms)Throughput (msg/s)Throughput (MB/s)
64 5,000 2,772.84 1,803.21 0.11
128 5,000 1,943.23 2,573.04 0.31
256 5,000 1,899.24 2,632.63 0.64
64K 1,000 3,020.98 331.02 20.69
128K 1,000 4,414.00 226.55 28.32
256K 1,000 20,911.61 47.82 11.96
512K 1,000 39,213.23 25.50 12.75
1MB 1,000 80,337.55 12.45 12.45

Load Test Result on HP ProLiant DL380 G7

Message SizeNumber of MessagesProcess Time (ms)Throughput (msg/s)Throughput (MB/s)
64 5,000 1,340.22 3,730.74 0.23
128 5,000 1,345.18 3,716.99 0.45
256 5,000 1,040.59 4,804.95 1.17
64K 1,000 2,178.69 458.99 28.69
128K 1,000 3,553.02 281.45 35.18
256K 1,000 7,490.54 133.50 33.38
512K 1,000 35,078.26 28.51 14.25
1MB 1,000 77,076.51 12.97 12.97

From this test result, we can conclude that HP ProLiant DL380 G7 performed better than Sun V490 under load.

Friday, February 08, 2013

Clojure Script to Convert Java GC log to CSV Format

My previous blog post showed how to use R to perform Java GC log analysis. Ajit Joglekar asked that I post the Clojure script. Here's the Clojure script that I used to convert GC log to CSV format. I have not tested this with a wide variety of GC logs but I do know that this script will not work for CMS or G1GC gc logs. Modify minor-gc-pattern and full-gc-pattern to match the gc log output as needed.


(use '[clojure.string :only (join)])  

; Match for pause time "0.1566980 secs]"
(def ^:constant pause-time "([\\d\\.]+) secs\\]")

; Match for Java heap space stat "524288K->32124K(2009792K)"
(def ^:constant space "(\\d+)K->(\\d+)K\\((\\d+)K\\)")

; Match for Execution stat "[Times: user=0.24 sys=0.06, real=0.16 secs]"
(def ^:constant exec-stat " \\[Times: user=([\\d\\.]+) sys=([\\d\\.]+), real=([\\d\\.]+) secs\\]")

; Example Minor GC entry   
; 212.785: [GC [PSYoungGen: 524288K->32124K(611648K)] 524288K->32124K(2009792K), 0.1566980 secs] [Times: user=0.24 sys=0.06, real=0.16 secs] 
; Define regex pattern to parse young gen GC event
(defn minor-gc-pattern []
  (let [timestamp  "([\\d\\.]+): \\[GC .*\\[PS.*: "
        young-gen   (str space "] ")
        heap        (str space ", ")]
    (re-pattern (str timestamp young-gen  heap pause-time exec-stat))))                

; Example Full GC entry   
;"43587.513: [Full GC (System) [PSYoungGen: 964K->0K(598912K)] [PSOldGen: 142673K->120674K(1398144K)] 143637K->120674K(1997056K) [PSPermGen: 82179K->82179K(147520K)], 0.7556570 secs] [Times: user=0.76 sys=0.00, real=0.77 secs]"
; Define the regex pattern to parse each line in gc log
(defn full-gc-pattern []
    (let [timestamp   "([\\d\\.]+): \\[Full.*"
          young-gen  (str ": " space "]")
          old-gen    (str " \\[\\w+: " space "\\] ")
          perm-gen   (str " \\[\\w+: "space "\\], ")
          heap       space]
         (re-pattern (str timestamp 
                          young-gen 
                          old-gen 
                          heap 
                          perm-gen 
                          pause-time 
                          exec-stat))))                
 
; Variable definitions (for both process-full-gc & process-minor-gc
;     ts - timestamp (in seconds)
;     ys - YoungGen space starting heap size (in KB)
;     ye - YoungGen space ending heap size (in KB)
;     ym - YoungGen space max heap size (in KB)
;     os - OldGen space starting heap size (in KB)
;     oe - OldGen space ending heap size (in KB)
;     om - OldGen space max heap size (in KB)
;     hs - Total heap space starting heap size (in KB)
;     he - Total heap space ending heap size (in KB)
;     hm - Total heap space max heap size (in KB)
;     pt - GC Pause Time (in seconds)
;     ps - PermGen space starting heap size (in KB)
;     pe - PermGen space ending heap size (in KB)
;     pm - PermGen space max heap size (in KB)
;     ut - User Time (in seconds)
;     kt - Kernel Time (in seconds)
;     rt - Real Time (in seconds)
(defn process-full-gc [entry]
  (let [[a ts ys ye ym os oe om hs he hm ps pe pm pt ut kt rt & e] entry]
    (join \, [ts "full" pt 
              ys ye ym 
              hs he hm 
              ut kt rt 
              os oe om 
              ps pe pm])))
   
(defn process-minor-gc [entry]
  (let [[a ts ys ye ym hs he hm pt ut kt rt & e] entry]
    (join \, [ts "minor" pt 
              ys ye ym 
              hs he hm 
              ut kt rt])))

(def headers (join \,  ["timestamp" "gc.type" "pause.time"
                        "young.start" "young.end" "young.max"
                        "heap.start" "heap.end" "heap.max"
                        "time.user" "time.sys" "time.real"
                        "old.start" "old.end" "old.max"
                        "perm.start" "perm.end" "perm.max"]))    
       

(defn process-gc-file [infile outfile]
  (let [gcdata (line-seq (clojure.java.io/reader (clojure.java.io/file infile)))]
    (with-open [w (clojure.java.io/writer outfile)]
      (let [writeln (fn [x] (.write w (str x "\n")))]
        (writeln headers)
        (doseq [line gcdata]
          (let [minor-gc (re-seq (minor-gc-pattern) line)
                full-gc  (re-seq (full-gc-pattern) line)]
            (when-not (nil? full-gc) 
              (writeln (process-full-gc (first full-gc))))
            (when-not (nil? minor-gc) 
              (writeln (process-minor-gc (first minor-gc))))))))))

;-----------------------------------------------------------------------   
; Convert Java GC log csv format
;-----------------------------------------------------------------------    
(process-gc-file "gc.log" "data.csv")

You can download the script directly here : gcanalysis.clj.

Monday, January 28, 2013

Checking Tibco EMS Queue Connected Users with Clojure

During our Tibco EMS infrastructure upgrade project, we had some rogue connections that just refused to disconnect from the old infrastructure and switch to the new Tibco EMS infrastructure. In order to track down these connections and evict those connections, I had to generate a report which would tell me which user, from which host is connected to particular queues in Tibco EMS. Here's the script that I wrote in Clojure that generates that report.

; Get destination users (and which host the user is connecting from) along with which queue/topic it is connected to.
(defn get-destination-users [server-url username password]
  (with-open [admin (TibjmsAdmin. server-url username password)]
    (let [connections  (.getConnections admin) 
          consumers    (.getConsumers admin)
          dest-userids (->> consumers
                         (map (fn [c] {:userid (.getConnectionID c) 
                                       :dest (.getDestinationName c)})))
          get-user (fn [dest-name id]
                     (let [conn (first (filter #(= id (.getID %)) connections))]
                       (if (nil? conn)
                         {:dest dest-name :user (str "Unknown user : " id) :host "unknown"}
                         {:dest dest-name :user (.getUserName conn) :host (.getHost conn)})))]
      (map (fn [x] (get-user (:dest x) (:userid x))) dest-userids))))

; Dump results in CSV format
(def results (get-destination-users server-url username password))   
(doseq [r (sort-by :user (set results))]
  (println  (str (:user r) "," (:host r) "," (:dest r))))

Monday, January 14, 2013

Who's Connecting to My Servers?

I have been reading through the book Clojure Programming. In the course of reading this book, I've looked for all sorts of opportunities to try applying Clojure at work. Most of the time, I've used Clojure to implement convenience scripts to help with my day job. I would typically use Clojure whenever I have to work with Java based platforms and F# for .NET based platforms. Occasionally, I would develop scripts that does not have any dependency and I could choose any language to implement. What would typically happen is that I would choose the programming language that I used last. This strategy, unfortunately, would typically end up biasing me toward one programming language and lately, it has been biasing me toward Clojure. After noticing this trend, I have decided to deliberately and consciously choose to implement in the less frequently used language so I don't become completely rusty in the other programming languages.

Recently, I had to opportunity to write a small script. I was managing an infrastructure upgrade and needed to know the downstream impact. It was an infrastructure component that that a lot of persistent inbound connections, but unfortunately, the inbound connections were neither monitored nor documented. One way to check the connections is ask the the network engineers to setup monitoring on the servers and collect the information on the incoming connections. Our network engineers are generally pretty busy and we hate to add to their existing workloads. However, we can effectively do the same thing by running netstat -an on each of the target servers and taking that output dump and parse that for incoming connections. We would do this over a period of time to try to capture most of the client connections.

The following Clojure script loads all the netstat dump output files and generate a list of all the hosts that are connected to the target servers:

(import '(java.net InetAddress))
(use '[clojure.string :only (join)])
(use '[clojure.java.io :as io])

; Load all the data from all *.data files in c:\work\servers folder
(def data (->> "c:\\work\\servers"
   (io/file)
   (file-seq)
   (map #(.getAbsolutePath %))
   (filter #(re-matches #".*\.data$" %))
   (map #(slurp %))
   (join " ")))

; Find all ip addresses in the netstat dump
; Perform hostname lookup, discard duplicates, sort the hostnames   
(def hosts (->> data
   (re-seq #"(\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3})\.\d+")
   (map #(second %))
   (set)
   (map #(.getCanonicalHostName (InetAddress/getByName %)))
   (sort)
   (join "\n")))

; Dump output to clients.out file
(spit "c:\\work\\servers\\results.out" hosts)

The above script runs with the assumption that all data fits into memory. However, if that becomes a problem, it is fairly trivial to sequentially read and process netstat dump one file at a time and combine the results to write to the output.

The F# version is similar to Clojure version. Grabbing the files from the folder is easier but the need to explicitly handle exceptions adds back the additional lines of code to be about on par with code verbosity of the Clojure version.

open System.IO
open System.Net
open System.Text.RegularExpressions

// Load all the data from all *.data files in c:\work\servers folder
let data = Directory.GetFiles(@"c:\work\servers","*.data")
           |> Seq.map File.ReadAllText
           |> String.concat " "

// Return hostname if it can be resolved
// otherwise return the ip address
let getHostEntry (ipaddress:string) =
    try
        Dns.GetHostEntry(ipaddress).HostName
    with
      | err -> ipaddress

// Find all ip addresses in the netstat dump
// Perform hostname lookup, discard duplicates, sort the hostnames
let hosts = Regex.Matches(data,@"(\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3})\.\d+")
            |> Seq.cast<Match>
            |> Seq.map (fun m -> m.Groups.[1].Value)
            |> Set.ofSeq
            |> Seq.map getHostEntry
            |> Seq.sort
            |> String.concat "\n"

File.WriteAllText(@"c:\work\servers\results.out",hosts)

Monday, June 11, 2012

Quick and easy way to monitor for memory leaks in Java application servers

I was part of a postmortem analysis team that performed a root cause analysis on why a particular node in a HA paired system failed and needed to provide recommendations to prevent future occurrences. Initial examination of the log showed that the node that failed had a permgen out of memory problem. Management team also wanted to know if there were memory leaks with the application.

Unfortunately, the Java application server was not configured with GC logging nor configured to perform a heap dump on out of memory error condition. Without the gc log data or the heap dump, it becomes harder to be able to explain why did the application server run out of permgen space at the time it failed. In the interest of collecting data and to prevent future node failures, we recommended that permgen space be increased and add the following startup parameters to the Java application server

    -XX:+PrintGCTimeStamps
    -XX:+PrintGCDetails
    -XX:+HeapDumpOnOutOfMemoryError
    -XX:HeapDumpPath=/myapp/heapdumps
    -Xloggc:/myapp/logs/gc.log

One problem that arose during the writeup of the analysis was how to answer the question whether the application had any memory leak issue. The application team could not reproduce the problem in non-production environment but wanted some assurance that the application did not have a memory leak problem with PermGen space or anywhere else in the heap. I proposed that we track heap usage after every Full GC over a period of time and see the heap usage trends over time. The heap usage trends should provide a good indication if the application has a problematic memory leak issue or not.

In order to do so, I wrote a small script to parse out only the heap and permgen space info after a full GC from the gc log. I wrote this script in Clojure as follows:

; Define the regex pattern to parse each line in gc log
(defn full-gc-pattern []
    (let [timestamp  "([\\d\\.]+): .* " 
          space      "(\\d+)K->(\\d+)K\\((\\d+)K.+ "
          new-gen    space
          old-gen    space
          perm-gen   space
          heap       "(\\d+)K->(\\d+)K\\((\\d+)K\\)\\], "
          exec-stat  "(\\d+\\.\\d+).*" ]
         (re-pattern (str timestamp new-gen old-gen perm-gen heap exec-stat))))                 

; We only want to dump timestamp, total heap and perm gen space
; after each full GC
; Variable definitions:
;     ts - timestamp (in seconds)
;     ys - YoungGen space starting heap size (in KB)
;     ye - YoungGen space ending heap size (in KB)
;     ym - YoungGen space max heap size (in KB)
;     os - OldGen space starting heap size (in KB)
;     oe - OldGen space ending heap size (in KB)
;     om - OldGen space max heap size (in KB)
;     hs - Total heap space starting heap size (in KB)
;     he - Total heap space ending heap size (in KB)
;     hm - Total heap space max heap size (in KB)
;     ps - PermGen space starting heap size (in KB)
;     pe - PermGen space ending heap size (in KB)
;     pm - PermGen space max heap size (in KB)
(defn process-full-gc [entry]
    (let [parsed (first (re-seq (full-gc-pattern) entry))
         [a ts ys ye ym os oe om hs he hm ps pe pm & e] parsed]
        (println (format "%s,%s,%s" ts he pe))))

; Load the gc log data
(def gcdata (line-seq (clojure.java.io/reader (clojure.java.io/file "gc.log"))))
        
; Process each full GC entry after filtering out minor GC entries          
(doseq [line (keep #(re-find #".*Full GC.*" %) gcdata)]
    (process-full-gc line))

Using this script against the gc log file, and taking that output and loading into a spreadsheet, I can plot and generate a trendline to see if memory is growing or not. Here's the heap usage plot over approximately a month.

As seen on the chart, the slope of the trendlines are small and negative, which is a good indicator that we do not have a memory leak problem with this application.

Monday, May 14, 2012

Working with Tibco EMS Message Topics using F# and Clojure

In my previous blog post, I showed how to connect to a Tibco EMS Queue with F# and Clojure to represent integration interoperability from .NET and Java platforms. Message queues are a way to implement the Request-Reply pattern, which is one of the many enterprise integration patterns described in the book Enterprise Integration Patterns. Another basic enterprise integration pattern is the Publish-Subscribe pattern, which can be implement via JMS topics. This blog post shows how to connect to JMS topic from .NET and Java using Tibco EMS as the JMS provider.

Here is the F# version:

#r @"C:\tibco\ems\6.3\bin\TIBCO.EMS.dll"

open System
open TIBCO.EMS

let serverUrl = "tcp://localhost:7222"
let producer = "producer"
let consumer = "consumer"
let password = "testpwd"
let topicName = "testTopic"


let subscribeToTopic serverUrl userid password topicName messageProcessor =
    async {
        let connection = (userid,password)
                         |> (new TopicConnectionFactory(serverUrl)).CreateTopicConnection
        let session = connection.CreateTopicSession(false,Session.AUTO_ACKNOWLEDGE)
        let topic = session.CreateTopic(topicName)
        let subscriber =  session.CreateSubscriber(topic)
        connection.Start()
        printf "Subscriber connected!\n"
        while true do
            try
                subscriber.Receive() |> messageProcessor
            with _ ->  ()
        connection.Close()
    }

let publishTopicMessages serverUrl  userid password topicName messages =
    let connection = (userid,password)
                     |> (new TopicConnectionFactory(serverUrl)).CreateTopicConnection
    let session = connection.CreateTopicSession(false,Session.AUTO_ACKNOWLEDGE)
    let topic = session.CreateTopic(topicName)
    let publisher = session.CreatePublisher(topic)
    connection.Start()

    messages
    |> Seq.iter (fun item -> session.CreateTextMessage(Text=item)
                             |> publisher.Send)
                             
    connection.Close()

// Just dump message to console for now
let myMessageProcessor (msg:Message) =
    msg.ToString() |> printf "%s\n"

let consumeMessageAsync = subscribeToTopic "tcp://localhost:7222" "consumer" "testpwd"


let produceMessages topicName messages = publishTopicMessages "tcp://localhost:7222" "producer" "testpwd" topicName messages 


// Asynchronously start the topic subscriber
Async.Start(consumeMessageAsync "testTopic" myMessageProcessor)


// Publish messages to the Tibco EMS topic
[ "Aslund"; "Barrayar"; "Beta Colony"; "Cetaganda"; "Escobar"; "Komarr"; "Marilac"; "Pol"; "Sergyar"; "Vervain"]
|> produceMessages "testTopic"


printf "Done!"

One thing to point out is that Tibco, unfortunately, did not implement IDisposable for it's Connection objects; perhaps in it's bid to stay faithful to the Java API. That design choice seems unfortunate to me in the sense that I no longer can leverage C#'s using keyword or F#'s use keyword to automatically close connection. I suppose it is fairly trivial to subclass the QueueConnection and TopicConnection class and add the IDisposable interface, but I feel that Tibco should have done this and developed the Tibco .NET API using idioms that are .NET specific.

Putting my rants aside, here is the equivalent Clojure code to connect to Tibco Topics:

(import '(java.util Enumeration)
        '(com.tibco.tibjms TibjmsTopicConnectionFactory)
        '(javax.jms Message JMSException  Session
                    Topic TopicConnectionFactory
                    TopicConnection TopicSession
                    TopicSubscriber))
                  
(def serverUrl "tcp://localhost:7222")
(def producer "producer")
(def consumer "consumer")
(def password "testpwd")
(def topicName "testTopic")

;------------------------------------------------------------------------------
; Subscribe to Topic asynchronously
;------------------------------------------------------------------------------
(defn subscribe-topic [server-url user password topic-name process-message]
    (future
        (with-open [connection (-> (TibjmsTopicConnectionFactory. server-url)
                                   (.createTopicConnection user password))]
            (let [session (.createTopicSession connection false Session/AUTO_ACKNOWLEDGE)
                  topic (.createTopic session  topic-name)]
                (with-open [subscriber (.createSubscriber session topic)]
                    (.start connection)
                    (loop []                       
                        (process-message (.receive subscriber))
                        (recur)))))))

;------------------------------------------------------------------------------
; Publishing to a Topic
;------------------------------------------------------------------------------
(defn publish-to-topic [server-url user password topic-name messages]
    (with-open [connection (-> (TibjmsTopicConnectionFactory. server-url)
                               (.createTopicConnection user password))]
        (let [session (.createTopicSession connection false Session/AUTO_ACKNOWLEDGE)
              topic (.createTopic session  topic-name)
              publisher (.createPublisher session topic)]
            (.start connection)
            (doseq [item messages]
                (let [message (.createTextMessage session)]
                    (.setText message item)
                    (.publish publisher message))))))
                    
                      
; Create function aliases with connection information embedded                    
(defn produce-messages [topic-name messages]
    (publish-to-topic "tcp://localhost:7222" "producer" "testpwd" topic-name messages))

(defn consume-messages [topic-name message-processor]
    (subscribe-topic "tcp://localhost:7222" "consumer" "testpwd" topic-name message-processor))

; Just dump messages to console for now
(defn my-message-processor [message]
    (println (.toString message)))
    
; Start subscribing messages asynchronously
(consume-messages "testTopic" my-message-processor)                            
    
; Publish to topic
(def my-messages '("alpha" "beta" "gamma" "delta"
                   "epsilon" "zeta" "eta" "theta"
                   "iota" "kappa" "lambda" "mu" "nu"
                   "xi", "omicron" "pi" "rho"
                   "signma" "tau" "upsilon" "phi",
                   "chi" "psi" "omega"))                    

(produce-messages  "testTopic"  my-messages)    

When I fire up both scripts, the messages published to the topic would be received by both the .NET and Java clients. With these scripts, I can easily swap out the message generators or message processors as needed for any future testing scenarios.

Monday, April 30, 2012

F#, Clojure and Message Queues on Tibco EMS

It looks like I will be getting much more hands on with Tibco EMS. Since the Tibco EMS system in use will have connections from both .NET platforms and Java platforms, I wanted to write some scripts to run some engineering tests on Tibco EMS. I decided to simulate .NET side connections with F# and Java side connections with Clojure. Taking the sample code from Tibco installation, I created the following F# script that sends messages to a queue from the sample C# code:

#r @"C:\tibco\ems\6.3\bin\TIBCO.EMS.dll"

open System
open TIBCO.EMS

let serverUrl = "tcp://localhost:7222"
let producer = "producer"
let consumer = "consumer"
let password = "testpwd"
let queueName = "testQueue"


let getQueueTextMessages serverUrl  userid password queueName messageProcessor =
    async {
        let connection = (userid,password)
                         |> (new QueueConnectionFactory(serverUrl)).CreateQueueConnection
        let session = connection.CreateQueueSession(false,Session.AUTO_ACKNOWLEDGE)
        let queue = session.CreateQueue(queueName)
        let receiver =  session.CreateReceiver(queue)
        connection.Start()
        printf "Queue connection established!"
        while true do
            try
                receiver.Receive() |> messageProcessor
            with _ ->  ()
    }


let sendQueueTextMessages serverUrl  userid password queueName messages =
    let connection = (userid,password)
                     |> (new QueueConnectionFactory(serverUrl)).CreateQueueConnection
    let session = connection.CreateQueueSession(false,Session.AUTO_ACKNOWLEDGE)
    let queue = session.CreateQueue(queueName)
    let sender = session.CreateSender(queue)
    connection.Start()

    messages
    |> Seq.iter (fun item -> session.CreateTextMessage(Text=item)
                             |> sender.Send)
                             
    connection.Close()



// Just dump message to console for now
let myMessageProcessor (msg:Message) =
    msg.ToString() |> printf "%s\n"


let consumeMessageAsync = getQueueTextMessages "tcp://localhost:7222" "consumer" "testpwd"
let produceMessages queueName messages = sendQueueTextMessages "tcp://localhost:7222" "producer" "testpwd" queueName messages 

// Start message consumer asynchronously
Async.Start(consumeMessageAsync "testQueue" myMessageProcessor)


// Send messages to the Tibco EMS   
[ "Aslund"; "Barrayar"; "Beta Colony"; "Cetaganda"; "Escobar"; "Komarr"; "Marilac"; "Pol"; "Sergyar"; "Vervain"]
|> produceMessages "testQueue"

The queue consumer is implemented asynchronously so it won't block executing subsequent statements. To test Tibco JMS from Java, here is the equivalent Clojure code:

(import '(java.util Enumeration)
        '(com.tibco.tibjms TibjmsQueueConnectionFactory)
        '(javax.jms Message JMSException  Session
                    Queue QueueBrowser 
                    QueueConnection QueueReceiver 
                    QueueSession QueueSender))
                  
(def serverUrl "tcp://localhost:7222")
(def producer "producer")
(def consumer "consumer")
(def password "testpwd")
(def queueName "testQueue")

; Consume Queue Text messages asynchronously
(defn get-queue-text-messages [server-url user password queue-name process-message]
    (future
        (with-open [connection (-> (TibjmsQueueConnectionFactory. server-url)
                                   (.createQueueConnection user password))]
            (let [session (.createQueueSession connection false Session/AUTO_ACKNOWLEDGE)
                  queue (.createQueue session  queue-name)]
                (with-open [receiver (.createReceiver session queue)]              
                    (.start connection)
                    (loop []                       
                        (process-message (.receive receiver))
                        (recur)))))))
                   
; Send multiple Text messages
(defn send-queue-text-messages [server-url user password queue-name messages]
    (with-open [connection (-> (TibjmsQueueConnectionFactory. server-url)
                               (.createQueueConnection user password))]
        (let [session (.createQueueSession connection false Session/AUTO_ACKNOWLEDGE)
              queue (.createQueue session  queue-name)
              sender (.createSender session queue)]
            (.start connection)
            (doseq [item messages]
                (let [message (.createTextMessage session)]
                    (.setText message item)
                    (.send sender message))))))


; Create function aliases with connection information embedded                    
(defn consume-messages [queue-name message-processor]
    (get-queue-text-messages  serverUrl producer password queue-name message-processor))

(defn produce-messages [queue-name messages]
    (send-queue-text-messages  serverUrl producer password queue-name messages))

; Just dump messages to console for now
(defn my-message-processor [message]
    (println (.toString message)))

    
; Start consuming messages asynchronously
(consume-messages "testQueue" my-message-processor)                            

; Send messages to queue
(def my-messages '("alpha" "beta" "gamma" "delta"
                   "epsilon" "zeta" "eta" "theta"
                   "iota" "kappa" "lambda" "mu" "nu"
                   "xi", "omicron" "pi" "rho"
                   "signma" "tau" "upsilon" "phi",
                   "chi" "psi" "omega"))                    

(produce-messages  "testQueue"  my-messages)    

With these scripts, I can easily swap in different message generators and message processors as needed for any testing purposes. When I fired up both these scripts up to the part where queue consumers are running in both F# and Clojure version and then send the messages, I could see that Tibco EMS send half the messages to my F# script and the other half to my Clojure script. Since both of these scripts run in REPL environment, I can easily adjust my level of testing as I get results.

Tuesday, March 06, 2012

Adventures in troubleshooting out of memory errors with Coherence cluster.

One day, an application team manager called me and said that their application caused an out of memory error condition in their Oracle Coherence cluster. This same code base ran in the old Coherence 3.1 environment for months without running into out of memory conditions and now is failing in the new Coherence 3.6 environment in matter of a few weeks on a regular basis. He said that he had heap dumps and logs and asked whether I could take a look at it and troubleshoot it.

Initially, I was skeptical about being able to help this team manager out. After all, I know almost nothing about their application code and in all practical terms, I had no previous development experience with Coherence with the exception that I read the book Oracle Coherence 3.5 by Aleksandar Seovic in the past. My previously participated in testing Coherence performance on VMware and that really did not require me to delve into the Coherence API at all.

Despite these misgivings, I decided to provide my support and told the application team manager that I'll try my best.

The system with problems was a multi-node Coherence cluster. When I took a look at the logs, all of them had these similar verbose GC output:

[GC [1 CMS-initial-mark: 1966076K(1966080K)] 2083794K(2084096K), 0.1923110 secs] [Times: user=0.18 sys=0.00, real=0.19 secs] 
[Full GC [CMS[CMS-concurrent-mark: 1.624/1.626 secs] [Times: user=3.22 sys=0.00, real=1.62 secs] 
 (concurrent mode failure): 1966079K->1966078K(1966080K), 6.6177340 secs] 2084093K->2084082K(2084096K), [CMS Perm : 13617K->13617K(23612K)], 6.6177900 secs] [Times: user=8.21 sys=0.00, real=6.62 secs] 
[Full GC [CMS: 1966078K->1966078K(1966080K), 4.1110330 secs] 2084093K->2084089K(2084096K), [CMS Perm : 13617K->13615K(23612K)], 4.1111070 secs] [Times: user=4.11 sys=0.00, real=4.11 secs] 
[Full GC [CMS: 1966078K->1966078K(1966080K), 4.2973090 secs] 2084092K->2084087K(2084096K), [CMS Perm : 13615K->13615K(23612K)], 4.2973630 secs] [Times: user=4.28 sys=0.00, real=4.30 secs] 
[Full GC [CMS: 1966078K->1966078K(1966080K), 4.1831450 secs] 2084093K->2084093K(2084096K), [CMS Perm : 13615K->13615K(23612K)], 4.1831970 secs] [Times: user=4.18 sys=0.00, real=4.18 secs] 
[Full GC [CMS: 1966078K->1966078K(1966080K), 4.2524850 secs] 2084093K->2084093K(2084096K), [CMS Perm : 13615K->13615K(23612K)], 4.2525380 secs] [Times: user=4.24 sys=0.00, real=4.25 secs] 
java.lang.OutOfMemoryError: Java heap space
Dumping heap to java_pid23607.hprof ...
Heap dump file created [2274434953 bytes in 28.968 secs]

This garbage collection log output tells me that they are using CMS for the JVM GC. The concurrent mode failure entries certainly grabbed my attention. Normally one would fix concurrent mode failures by tuning the CMS initiation occupancy fraction via -XX:CMSInitiatingOccupancyFraction flag, but in this case, looking that the heap numbers in the lines labeled "Full GC" showed that GC could not clean up any memory at all. So this problem could not be solved by GC tuning. By the way, for a great book on tuning garbage collection, I would recommend Charlie Hunt's book Java Performance.

My next step was to take a look a the heap. The heap was slightly over 2 GB, which was expected since Coherence cluster node was each configured with a 2GB heap. Well, that presented a problem for me because I'm still mainly working on a 32-bit Windows laptop. I needed to find a 64-bit system with preferably 4 GB of ram or more to look at this. Once I was able to get such a machine and fired up Eclipse Memory Analyzer Tool (MAT). Once I looked at the heap, it was pretty obvious what was the biggest memory offender. The biggest memory offender was a top level hashmap chewing up 1.6 GB of memory. Delving further into that hash map structure, it reveals that Coherence caching structure is a hash of hashes. Looking at the hashes, I notice that there were over 2000+ items in the top level hash. That would imply that there were over 2000+ caches in the Coherence cluster. Studying each individual cache, I would notice cache names like

  • alpha-FEB-21
  • alpha-FEB-22
  • alpha-FEB-23
  • alpha-FEB-24
  • beta-FEB-21
  • beta-FEB-22
  • beta-FEB-23
  • beta-FEB-24

and so forth. I ask the application team manager whether he expected to have this many caches in the cluster. The application team manager said no; he expected a much smaller set of caches. The application normally destroy caches older than 2 days. The developers provided me their code related to the creation and destruction of caches and I saw the following lines of code and it seems pretty innocuous:

    public static void destroyCache(String name) {
        Collection listOfCacheNames = getListOfCacheNames(name, false);
        Iterator iterator = listOfCacheNames.iterator();
        while (iterator.hasNext()) {
            String name = (String) iterator.next();
            NamedCache namedCache = CacheFactory.getCache(name);
            namedCache.destroy();
        }
    }

I went back to the memory analyzer tool and performed a GC to root analysis and saw the top level object that's holding onto this heap as:

com.tangosol.coherence.component.net.Cluster$IpMonitor @ 0x77ff4b18 

with the label "Busy Monitor" next to it. This line item seems to suggest that there's a monitor lock on this cache. Looking at the Coherence API documentation, I see the following entry:


destroy

void destroy()
Release and destroy this instance of NamedCache.

Warning: This method is used to completely destroy the specified cache across the cluster. All references in the entire cluster to this cache will be invalidated, the cached data will be cleared, and all internal resources will be released.

Caches should be destroyed by the same mechansim in which they were obtained. For example:

  • new Cache() - cache.destroy()
  • CacheFactory.getCache() - CacheFactory.destroyCache()
  • ConfigurableCacheFactory.ensureCache() - ConfigurableCacheFactory.destroyCache()
Except for the case where the application code expicitly allocated the cache, this method should not be called by application code.

Looking at this documentation, we initially thought that since the cache was obtained via CacheFactory and therefore should be destroyed via CacheFactory ergo CacheFactory had a monitor lock on the underlying collections. The code provided by the developers used one mechanism to create the cache and another mechanism to destroy the cache so we presume that was the problem. So I implemented a test script to test out that theory and surprisingly, even destroying via CacheFactory, I still encounter out of memory issues. Only by clearing the cache before destroying the cache was I able to avoid out of memory errors. Here's the script that I developed in Clojure to test my theories:

(import '(org.apache.commons.lang3 RandomStringUtils) 
        '(java.math BigInteger)
        '(java.util Random Date HashMap)
        '(com.tangosol.net NamedCache CacheFactory CacheService Cluster))

(defn random-text [] (RandomStringUtils/randomAlphanumeric 1048576))
(defn random-key [] (RandomStringUtils/randomAlphanumeric 12))
        
(CacheFactory/ensureCluster)
(def buffer (new HashMap))

(defn print-horizontal-line [c] (println (apply str (repeat 80 c))))

(def caches '("alpha" "beta" "gamma" "delta" 
              "epsilon" "zeta" "eta" "theta" 
              "iota" "kappa" "lambda" "mu" "nu"
              "xi", "omicron" "pi" "rho"
              "signma" "tau" "upsilon" "phi", 
              "chi" "psi" "omega"))

(defn load-cache [cache-name n]
    (let [cache (CacheFactory/getCache cache-name)]
         (print-horizontal-line  "=")
         (println "Creating cache : " cache-name)
         (print-horizontal-line  "=")         
         (.clear buffer)
         (dotimes [_ n] (.put buffer (random-key) (random-text)))
         (.putAll cache buffer)))


(defn recreate-oom-problem [cache-name]
    (let [cache (CacheFactory/getCache cache-name)]
         (load-cache cache-name 200)
         (.destroy cache)))
         
(defn try-fix-oom-1 [cache-name]
    (let [cache (CacheFactory/getCache cache-name)]
         (load-cache cache-name 200)
         (CacheFactory/destroyCache cache)))

(defn try-fix-oom-2 [cache-name]
    (let [cache (CacheFactory/getCache cache-name)]
         (load-cache cache-name 200)
         (.clear cache)
         (CacheFactory/destroyCache cache)))
         
; Test run recreation of original problem.  Was able to reproduce OOM issues         ; 
(doseq [cache caches] (recreate-oom-problem cache))

; Surprise! Still have OOM issues
(doseq [cache caches] (try-fix-oom-1 cache))

; No longer have OOM issues, but memory is still leaking (slowly)
(doseq [cache caches] (try-fix-oom-2 cache))
         

However, I still suspect memory leaks, it's just that my memory leak is a lot smaller now. To verify that I had a memory leak, I would run my Clojure test script and the deliberately create and fill a cache without clearing it. I then forced a full garbage collection followed by a heap dump. In memory analyzer tool, I would look up the cache that I did not clear, and list all the incoming references. Then I would look for a HashMap in the incoming references and select one of those and check for outgoing references. And in that outgoing references, I could see that the key contains the name of a cache that I had called CacheFactory.destroyCache() on and the retained heap sizes range anywhere from 24 to 160 with the sizes that seems proportional to the size of the cache name.

In conclusion, it would seem Oracle Coherence does have a memory leak issues with the cache creation and destruction process. If we clear the cache before destroying the cache, I suspect it would be a long time before the memory leak is even noticeable by this particular application.

To verify that this leak did not exist in the older 3.1 version, we ran this test code on and and was unable to reproduce the out of memory errors. We also tested this against Oracle Coherence 3.7.1 and was unable to reproduce the out of memory error. So, it looks like that this memory error is specific to Oracle Coherence 3.6 only.

Throughout this entire process, I thought that the secret sauce that enabled me to quickly learn Coherence, reproduce and troubleshoot the Coherence out of memory problem was Clojure. Clojure allowed me to interactively manipulate Coherence clusters and explore the API, which would have been a lot slower if I had to go through the normal edit-compile-run cycle with plain old Java.

Wednesday, January 11, 2012

Testing Coherence with Clojure

A developer came to me the other day asking for help in diagnosing some issues with their application and the interaction with Oracle's Coherence product. I wanted to write some testing harness to quickly test some Coherence configuration and gave some thought about how I would go and try to replicate the issues that the application had. I wanted a REPL environment so that I can interactive manipulate the Coherence API and dump outputs on demand. I decided to use Clojure to experiment with Coherence, although I could have used JRuby, Jython, Groovy or Scala. From purely a familiarity perspective, I would rank my usage of these listed languages in the order of Ruby first, Python second, Groovy third, Clojure fourth and Scala last. But for some unknown, deep-seated and probably emotional reasons, I like Clojure more and relish the opportunity to use it. One of the first thing I tried with Clojure and Coherence is to perform a timing test on adding data to a 2 node distributed cache in serial vs concurrent mode. Here's the example Clojure script:

(import '(org.apache.commons.lang3 RandomStringUtils) 
        '(java.math BigInteger)
        '(java.util Random Date HashMap)
        '(com.tangosol.net NamedCache CacheFactory CacheService Cluster))

(CacheFactory/ensureCluster)
(def cache (CacheFactory/getCache "sandbox"))

(defn random-text [] (RandomStringUtils/randomAlphanumeric 1048576))
(defn random-key [] (RandomStringUtils/randomAlphanumeric 12))


; Testing serial puts
(new Date)
(dotimes [_ 200] (.put cache (random-key) (random-text)))
(new Date)

; Testing concurrent puts
(def buffer (new HashMap))
(new Date)
(dotimes [_ 200] (.put buffer (random-key) (random-text)))
(.putAll cache buffer)
(new Date)

Running this code showed a 2x gain in speed of data load. On one of the Coherence nodes, I had JVisualVM connected to it and watched the realtime GC behaviors with VisualGC. It has been fascinating to watch the behaviorial differences between serial vs parallel data load and the memory activities of the Coherence node when my Clojure script was idle. I hope to conduct more tests in the future looking at GC behaviors and leverage my Clojure script as load driver in my testing efforts and assist me in GC tuning of Coherence instances.

Monday, August 30, 2010

Exploring Java AES encryption algorithm with Clojure

Encryption is one of those library that I use infrequently. It seems that every time I need to work with some encryption algorithm, it passes memory expiration date of my last implementation usage of some encryption algorithm. Recently, I had to implement some password encryption tool in Java and again, I had to revisit Java’s encryption library. In the past, for exploring unfamiliar libraries, I would normally write some code sprinkled with print statements, compile it and run the compiled code and adjust my code afterwards. But in the past couple years, I’ve become a big fan of REPL development environment for purposes of library exploration. In the .NET environment, I would use F# Interactive, Ruby & Python have their own REPL environment. For Java, I could have picked either Groovy or Clojure. I’ve scripted Groovy in the past and certainly liked the shorthand expressions to a lot of the Java code. However, I have heard a lot of positive blog posts about Clojure such that I wanted to try it out.

I never had the chance to program in Lisp in the past. But I have read some of Paul Graham’s articles which seems to make Lisp a programmer’s great secret weapon in becoming a better than average programmer. So I figure Clojure can get my feet wet with another variant of Lisp and help me accomplish my day job goals at the same time. The hardest part about learning Clojure was learning the API. The program structure was not hard to figure out. I used an RPN based HP 11C calculator while in college and Clojure program structure reminds me a lot of working with RPN calculators (albeit with a lot more parenthesis). Once I got the hang of the basic Clojure syntax, it became a great environment to work in. I would define a function, run the code and then interactively explore properties or call methods. After exploring in Clojure, it became fairly trivial task to translate that to the Java code.

While it was certainly fun to work in Clojure, I'm not sure I can substantiate Paul Graham's claim that knowing Lisp makes you a better than average programmer. Maybe it's more of the concept that a polyglot programmer generally makes a better programmer simply because the programmer that goes and learns multiple programming languages has a self driven desire to become a better programmer and therefore becomes a better programmer. I'll leave that discussion to the programmer philosophers out there and return to the more pragmmatic code construction.

Here’s the prototype AES encryption code in Clojure:

(import (javax.crypto KeyGenerator SecretKey Cipher))
(import (javax.crypto.spec SecretKeySpec))
(import (java.io File FileOutputStream DataInputStream FileInputStream))
(import (java.util Properties))
(import (org.apache.commons.codec.binary Base64))

(def msg "Hello encryption world!")
(defn encode-base64 [raw] (. (new Base64) encode raw))
(defn decode-base64 [coded] (. (new Base64) decode coded))

(def aes (. KeyGenerator getInstance "AES"))
(def cipher (. Cipher getInstance "AES"))
(def encrypt (. Cipher ENCRYPT_MODE))
(def decrypt (. Cipher DECRYPT_MODE))

(defn writekey [rawkey filename]     
    ( let [out (new FileOutputStream (new File filename))]
          (do (. out write rawkey)
              (. out close))))
    
(defn readkey [filename]
    (let [file (new File filename)
          rawkey (byte-array (. file length))
          in  (new DataInputStream (new FileInputStream file))]
          (do (. in readFully rawkey)
              (. in close)
              rawkey
          )))
          
(defn get-propfile [filename]   
    (let [prop (new Properties)]
        (do (. prop load (new FileInputStream filename)))
        prop))
    

(defn genkey [keygen] 
    (do (. keygen init  128)
        (. (. keygen generateKey ) getEncoded)
    )
)    
    
(defn do-encrypt [rawkey plaintext]
    (let [cipher (. Cipher getInstance "AES")]
        (do (. cipher init encrypt (new SecretKeySpec rawkey "AES"))
            (. cipher doFinal (. plaintext getBytes)))))
    
(defn do-decrypt [rawkey ciphertext]
    (let [cipher (. Cipher getInstance "AES")]
        (do (. cipher init  decrypt (new SecretKeySpec rawkey "AES"))
            (new String(. cipher doFinal ciphertext)))))
            
            
(defn get-password [key rawkey filename]
    (let [ props (get-propfile filename)
           coded (. props getProperty key)
           cipher (decode-base64 coded)]
         (do (do-decrypt rawkey cipher))))
         
(comment "Example usage"
(get-password "jms" (readkey "test.key") "data.out")
)