chia7712 commented on code in PR #22454:
URL: https://github.com/apache/kafka/pull/22454#discussion_r3474170977
##########
server/src/main/java/org/apache/kafka/server/metrics/LinuxIoMetricsCollector.java:
##########
@@ -35,36 +35,98 @@ public class LinuxIoMetricsCollector {
private static final Logger LOG =
LoggerFactory.getLogger(LinuxIoMetricsCollector.class);
private static final String READ_BYTES_PREFIX = "read_bytes: ";
private static final String WRITE_BYTES_PREFIX = "write_bytes: ";
+ private static final String RCHAR_PREFIX = "rchar: ";
+ private static final String WCHAR_PREFIX = "wchar: ";
+ private static final String SYSCR_PREFIX = "syscr: ";
+ private static final String SYSCW_PREFIX = "syscw: ";
+ private static final String CANCELLED_WRITE_BYTES_PREFIX =
"cancelled_write_bytes: ";
private final Time time;
private final Path path;
- private long lastUpdateMs = -1L;
- private long cachedReadBytes = 0L;
- private long cachedWriteBytes = 0L;
+ private volatile long lastUpdateMs = -1L;
+ private volatile long cachedReadBytes = 0L;
+ private volatile long cachedWriteBytes = 0L;
+ private volatile long cachedRchar = 0L;
+ private volatile long cachedWchar = 0L;
+ private volatile long cachedSyscr = 0L;
+ private volatile long cachedSyscw = 0L;
+ private volatile long cachedCancelledWriteBytes = 0L;
public LinuxIoMetricsCollector(String procRoot, Time time) {
this.time = time;
path = Paths.get(procRoot, "self", "io");
}
public long readBytes() {
- synchronized (this) {
- long curMs = time.milliseconds();
- if (curMs != lastUpdateMs) {
- updateValues(curMs);
- }
- return cachedReadBytes;
- }
+ refreshIfStale();
+ return cachedReadBytes;
}
public long writeBytes() {
- synchronized (this) {
- long curMs = time.milliseconds();
- if (curMs != lastUpdateMs) {
- updateValues(curMs);
+ refreshIfStale();
+ return cachedWriteBytes;
+ }
+
+ /**
+ * Returns the total number of characters read (includes cached reads).
+ * This value represents all read operations, including those satisfied by
the page cache.
+ */
+ public long rchar() {
+ refreshIfStale();
+ return cachedRchar;
+ }
+
+ /**
+ * Returns the total number of characters written (includes cached writes).
+ * This value represents all write operations, including those that may
not have reached disk.
+ */
+ public long wchar() {
+ refreshIfStale();
+ return cachedWchar;
+ }
+
+ /**
+ * Returns the number of read system calls.
+ * This metric helps identify I/O patterns and syscall overhead.
+ */
+ public long syscr() {
+ refreshIfStale();
+ return cachedSyscr;
+ }
+
+ /**
+ * Returns the number of write system calls.
+ * This metric helps identify I/O patterns and syscall overhead.
+ */
+ public long syscw() {
+ refreshIfStale();
+ return cachedSyscw;
+ }
+
+ /**
+ * Returns the number of bytes that were cancelled before being written.
+ * This can occur when a write is truncated or cancelled.
+ */
+ public long cancelledWriteBytes() {
+ refreshIfStale();
+ return cachedCancelledWriteBytes;
+ }
+
+ /**
+ * Refreshes all cached values from /proc/self/io if more than one
millisecond has elapsed
+ * since the last refresh. Uses double-checked locking so getters return
without acquiring
+ * the monitor in the common (cached) case — across all 7 gauges in a
single JMX scrape
+ * the lock is taken at most once per ms-window, and zero times in steady
state.
+ */
+ private void refreshIfStale() {
+ long curMs = time.milliseconds();
+ if (curMs != lastUpdateMs) {
+ synchronized (this) {
Review Comment:
How about using CAS instead of locking?
```java
private void refreshIfStale() {
long currentMs = time.milliseconds();
if (currentMs != lastUpdateMs) {
if (isUpdating.compareAndSet(false, true)) {
try {
currentMs = time.milliseconds(); // double check with
the latest time
if (currentMs != lastUpdateMs) {
updateValues(currentMs);
}
} finally {
isUpdating.set(false);
}
}
}
}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
