Changeset 4166

Timestamp:

2010-07-20 13:38:32 (3 years ago)

Author:

bruno

Message:

Small metrics change, check thread interrupted state before processing each message, do the thread wait completely following the pattern from wait javadoc.

File:

• trunk/rowlog/impl/src/main/java/org/lilycms/rowlog/impl/RowLogProcessorImpl.java (modified) (6 diffs)

trunk/rowlog/impl/src/main/java/org/lilycms/rowlog/impl/RowLogProcessorImpl.java

@@ -308 +308 @@
                 try {
                     List<RowLogMessage> messages = shard.next(consumer.getId());
-                    metrics.incScanCount();
+                    metrics.setScannedMessages(messages != null ? messages.size() : 0);
                     if (messages != null && !messages.isEmpty()) {
-                        metrics.incMessageCount();
                         for (RowLogMessage message : messages) {
+                            if (isInterrupted())
+                                return;
+
+                            metrics.incMessageCount();
                             byte[] lock = rowLog.lockMessage(message, consumer.getId());
                             if (lock != null) {
@@ -326 +329 @@
                         try {
                             long timeout = 5000;
-                            if (lastWakeup + timeout < System.currentTimeMillis()) {
-                                synchronized (this) {
+                            synchronized (this) {
+                                while (lastWakeup + timeout < System.currentTimeMillis()) {
                                     wait(timeout);
                                 }
@@ -344 +347 @@
         private class ProcessorMetrics implements Updater {
             private int scanCount = 0;
+            private long scannedMessageCount = 0;
             private int messageCount = 0;
             private int successCount = 0;
@@ -358 +362 @@
             public synchronized void doUpdates(MetricsContext unused) {
                 record.setMetric("scanCount", scanCount);
+                record.setMetric("messagesPerScan", scanCount > 0 ? scannedMessageCount / scanCount : 0f);
                 record.setMetric("messageCount", messageCount);
                 record.setMetric("successCount", successCount);
@@ -365 +370 @@
 
                 scanCount = 0;
+                scannedMessageCount = 0;
                 messageCount = 0;
                 successCount = 0;
@@ -371 +377 @@
             }
 
-            synchronized void incScanCount() {
+            synchronized void setScannedMessages(int read) {
                 scanCount++;
+                scannedMessageCount += read;
             }