You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@uniffle.apache.org by "lifeSo (via GitHub)" <gi...@apache.org> on 2023/10/28 11:52:45 UTC
[PR] [Improvement]: Change transportTime and processTime summary to Thread Pool Instead of block [incubator-uniffle]
lifeSo opened a new pull request, #1272:
URL: https://github.com/apache/incubator-uniffle/pull/1272
Change transportTime and processTime summary to Thread Pool Instead of block
Closes #1271
### What changes were proposed in this pull request?
Change transportTime and processTime summary to Thread Pool Instead of block
### Why are the changes needed?
Not block GRPC Thread
Fix: #1271
### Does this PR introduce _any_ user-facing change?
No.
### How was this patch tested?
No
--
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: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
Re: [PR] [#1271] improvement(server): change transportTime and processTime summary to Thread Pool Instead of block [incubator-uniffle]
Posted by "jerqi (via GitHub)" <gi...@apache.org>.
jerqi commented on code in PR #1272:
URL: https://github.com/apache/incubator-uniffle/pull/1272#discussion_r1375723397
##########
common/src/main/java/org/apache/uniffle/common/metrics/RPCMetrics.java:
##########
@@ -18,27 +18,45 @@
package org.apache.uniffle.common.metrics;
import java.util.Map;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
import com.google.common.collect.Maps;
+import com.google.common.collect.Queues;
import io.prometheus.client.CollectorRegistry;
import io.prometheus.client.Counter;
import io.prometheus.client.Gauge;
import io.prometheus.client.Summary;
import org.apache.uniffle.common.util.Constants;
import org.apache.uniffle.common.util.JavaUtils;
+import org.apache.uniffle.common.util.ThreadUtils;
public abstract class RPCMetrics {
protected boolean isRegistered = false;
protected Map<String, Counter.Child> counterMap = JavaUtils.newConcurrentMap();
protected Map<String, Gauge.Child> gaugeMap = JavaUtils.newConcurrentMap();
protected Map<String, Summary.Child> transportTimeSummaryMap = JavaUtils.newConcurrentMap();
protected Map<String, Summary.Child> processTimeSummaryMap = JavaUtils.newConcurrentMap();
+ private static final int WAIT_QUEUE_SIZE = 1000;
+ private final ExecutorService summaryObservePool;
protected MetricsManager metricsManager;
protected String tags;
public RPCMetrics(String tags) {
this.tags = tags;
+ BlockingQueue<Runnable> waitQueue = Queues.newLinkedBlockingQueue(WAIT_QUEUE_SIZE);
+ this.summaryObservePool =
+ new ThreadPoolExecutor(
+ 2,
+ 10,
+ 60,
+ TimeUnit.SECONDS,
+ waitQueue,
+ ThreadUtils.getThreadFactory("SummaryObserveThreadPool"),
+ new ThreadPoolExecutor.DiscardPolicy());
Review Comment:
Why do we choose to discard? It may cause the loss of metrics.
--
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: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
Re: [PR] [#1271] improvement(server): change transportTime and processTime summary to Thread Pool Instead of block [incubator-uniffle]
Posted by "lifeSo (via GitHub)" <gi...@apache.org>.
lifeSo commented on code in PR #1272:
URL: https://github.com/apache/incubator-uniffle/pull/1272#discussion_r1375708314
##########
common/src/main/java/org/apache/uniffle/common/metrics/RPCMetrics.java:
##########
@@ -18,27 +18,44 @@
package org.apache.uniffle.common.metrics;
import java.util.Map;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
import com.google.common.collect.Maps;
+import com.google.common.collect.Queues;
import io.prometheus.client.CollectorRegistry;
import io.prometheus.client.Counter;
import io.prometheus.client.Gauge;
import io.prometheus.client.Summary;
import org.apache.uniffle.common.util.Constants;
import org.apache.uniffle.common.util.JavaUtils;
+import org.apache.uniffle.common.util.ThreadUtils;
public abstract class RPCMetrics {
protected boolean isRegistered = false;
protected Map<String, Counter.Child> counterMap = JavaUtils.newConcurrentMap();
protected Map<String, Gauge.Child> gaugeMap = JavaUtils.newConcurrentMap();
protected Map<String, Summary.Child> transportTimeSummaryMap = JavaUtils.newConcurrentMap();
protected Map<String, Summary.Child> processTimeSummaryMap = JavaUtils.newConcurrentMap();
+ private final ExecutorService summaryObservePool;
protected MetricsManager metricsManager;
protected String tags;
public RPCMetrics(String tags) {
this.tags = tags;
+ BlockingQueue<Runnable> waitQueue = Queues.newLinkedBlockingQueue(1000);
Review Comment:
ok, I modified it .
--
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: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
Re: [PR] [#1271] improvement(server): change transportTime and processTime summary to Thread Pool Instead of block [incubator-uniffle]
Posted by "lifeSo (via GitHub)" <gi...@apache.org>.
lifeSo commented on code in PR #1272:
URL: https://github.com/apache/incubator-uniffle/pull/1272#discussion_r1375729098
##########
common/src/main/java/org/apache/uniffle/common/metrics/RPCMetrics.java:
##########
@@ -18,27 +18,45 @@
package org.apache.uniffle.common.metrics;
import java.util.Map;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
import com.google.common.collect.Maps;
+import com.google.common.collect.Queues;
import io.prometheus.client.CollectorRegistry;
import io.prometheus.client.Counter;
import io.prometheus.client.Gauge;
import io.prometheus.client.Summary;
import org.apache.uniffle.common.util.Constants;
import org.apache.uniffle.common.util.JavaUtils;
+import org.apache.uniffle.common.util.ThreadUtils;
public abstract class RPCMetrics {
protected boolean isRegistered = false;
protected Map<String, Counter.Child> counterMap = JavaUtils.newConcurrentMap();
protected Map<String, Gauge.Child> gaugeMap = JavaUtils.newConcurrentMap();
protected Map<String, Summary.Child> transportTimeSummaryMap = JavaUtils.newConcurrentMap();
protected Map<String, Summary.Child> processTimeSummaryMap = JavaUtils.newConcurrentMap();
+ private static final int WAIT_QUEUE_SIZE = 1000;
+ private final ExecutorService summaryObservePool;
protected MetricsManager metricsManager;
protected String tags;
public RPCMetrics(String tags) {
this.tags = tags;
+ BlockingQueue<Runnable> waitQueue = Queues.newLinkedBlockingQueue(WAIT_QUEUE_SIZE);
+ this.summaryObservePool =
+ new ThreadPoolExecutor(
+ 2,
+ 10,
+ 60,
+ TimeUnit.SECONDS,
+ waitQueue,
+ ThreadUtils.getThreadFactory("SummaryObserveThreadPool"),
+ new ThreadPoolExecutor.DiscardPolicy());
Review Comment:
> Why do we choose to discard? It may cause the loss of metrics.
The metric calc is more slow than the GRPC thread produce metric.
And discard will not effect the server function, so we choose to discard it.
Or Dalao have good idea ?
--
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: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
Re: [PR] [#1271] improvement(server): change transportTime and processTime summary to Thread Pool Instead of block [incubator-uniffle]
Posted by "lifeSo (via GitHub)" <gi...@apache.org>.
lifeSo commented on PR #1272:
URL: https://github.com/apache/incubator-uniffle/pull/1272#issuecomment-1784554644
> > > What will it happen if we don't have this pr?
> >
> >
> > GRPC Thread will block on the metric calc. Here is the jstack from PRD, We find many GRPC Thread blocks. So, if we don't have this pr, the GRPC Thread will block.
> > 
>
> Will it cause many requests timeout?
It will cause the GRPC block, then GRPC request queue come with many request task.
Some client will read shuffle data fail
--
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: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
Re: [PR] [#1271] improvement(server): change transportTime and processTime summary to Thread Pool Instead of block [incubator-uniffle]
Posted by "lifeSo (via GitHub)" <gi...@apache.org>.
lifeSo commented on PR #1272:
URL: https://github.com/apache/incubator-uniffle/pull/1272#issuecomment-1784526157
> What will it happen if we don't have this pr?
GRPC Thread will block on the metric calc.
Here is the jstack from PRD, We find many GRPC Thread blocks.
So, if we don't have this pr, the GRPC Thread will block.
--
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: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
Re: [PR] [#1271] improvement(server): change transportTime and processTime summary to Thread Pool Instead of block [incubator-uniffle]
Posted by "lifeSo (via GitHub)" <gi...@apache.org>.
lifeSo commented on code in PR #1272:
URL: https://github.com/apache/incubator-uniffle/pull/1272#discussion_r1377075699
##########
common/src/main/java/org/apache/uniffle/common/metrics/RPCMetrics.java:
##########
@@ -18,27 +18,45 @@
package org.apache.uniffle.common.metrics;
import java.util.Map;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
import com.google.common.collect.Maps;
+import com.google.common.collect.Queues;
import io.prometheus.client.CollectorRegistry;
import io.prometheus.client.Counter;
import io.prometheus.client.Gauge;
import io.prometheus.client.Summary;
import org.apache.uniffle.common.util.Constants;
import org.apache.uniffle.common.util.JavaUtils;
+import org.apache.uniffle.common.util.ThreadUtils;
public abstract class RPCMetrics {
protected boolean isRegistered = false;
protected Map<String, Counter.Child> counterMap = JavaUtils.newConcurrentMap();
protected Map<String, Gauge.Child> gaugeMap = JavaUtils.newConcurrentMap();
protected Map<String, Summary.Child> transportTimeSummaryMap = JavaUtils.newConcurrentMap();
protected Map<String, Summary.Child> processTimeSummaryMap = JavaUtils.newConcurrentMap();
+ private static final int WAIT_QUEUE_SIZE = 1000;
Review Comment:
Ok, I changed it to using config
--
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: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
Re: [PR] [#1271] improvement(server): change transportTime and processTime summary to Thread Pool Instead of block [incubator-uniffle]
Posted by "jerqi (via GitHub)" <gi...@apache.org>.
jerqi merged PR #1272:
URL: https://github.com/apache/incubator-uniffle/pull/1272
--
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: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
Re: [PR] [#1271] improvement(server): change transportTime and processTime summary to Thread Pool Instead of block [incubator-uniffle]
Posted by "jerqi (via GitHub)" <gi...@apache.org>.
jerqi commented on PR #1272:
URL: https://github.com/apache/incubator-uniffle/pull/1272#issuecomment-1784540861
> > What will it happen if we don't have this pr?
>
> GRPC Thread will block on the metric calc. Here is the jstack from PRD, We find many GRPC Thread blocks. So, if we don't have this pr, the GRPC Thread will block.
>
> 
Will it cause many requests timeout?
--
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: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
Re: [PR] [Improvement]: Change transportTime and processTime summary to Thread Pool Instead of block [incubator-uniffle]
Posted by "jerqi (via GitHub)" <gi...@apache.org>.
jerqi commented on PR #1272:
URL: https://github.com/apache/incubator-uniffle/pull/1272#issuecomment-1784106409
@lifeSo Could you change your title? We should use `[#xxxx]` or `[MINOR]` as the start of our title.
--
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: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
Re: [PR] [Improvement]: Change transportTime and processTime summary to Thread Pool Instead of block [incubator-uniffle]
Posted by "jerqi (via GitHub)" <gi...@apache.org>.
jerqi commented on PR #1272:
URL: https://github.com/apache/incubator-uniffle/pull/1272#issuecomment-1784105648
What will it happen if we don't have this pr?
--
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: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
Re: [PR] [#1271] improvement(server): change transportTime and processTime summary to Thread Pool Instead of block [incubator-uniffle]
Posted by "lifeSo (via GitHub)" <gi...@apache.org>.
lifeSo commented on code in PR #1272:
URL: https://github.com/apache/incubator-uniffle/pull/1272#discussion_r1377000450
##########
common/src/main/java/org/apache/uniffle/common/metrics/RPCMetrics.java:
##########
@@ -18,27 +18,45 @@
package org.apache.uniffle.common.metrics;
import java.util.Map;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
import com.google.common.collect.Maps;
+import com.google.common.collect.Queues;
import io.prometheus.client.CollectorRegistry;
import io.prometheus.client.Counter;
import io.prometheus.client.Gauge;
import io.prometheus.client.Summary;
import org.apache.uniffle.common.util.Constants;
import org.apache.uniffle.common.util.JavaUtils;
+import org.apache.uniffle.common.util.ThreadUtils;
public abstract class RPCMetrics {
protected boolean isRegistered = false;
protected Map<String, Counter.Child> counterMap = JavaUtils.newConcurrentMap();
protected Map<String, Gauge.Child> gaugeMap = JavaUtils.newConcurrentMap();
protected Map<String, Summary.Child> transportTimeSummaryMap = JavaUtils.newConcurrentMap();
protected Map<String, Summary.Child> processTimeSummaryMap = JavaUtils.newConcurrentMap();
+ private static final int WAIT_QUEUE_SIZE = 1000;
+ private final ExecutorService summaryObservePool;
protected MetricsManager metricsManager;
protected String tags;
public RPCMetrics(String tags) {
this.tags = tags;
+ BlockingQueue<Runnable> waitQueue = Queues.newLinkedBlockingQueue(WAIT_QUEUE_SIZE);
+ this.summaryObservePool =
+ new ThreadPoolExecutor(
+ 2,
Review Comment:
Ok, I changed it.
--
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: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
Re: [PR] [Improvement]: Change transportTime and processTime summary to Thread Pool Instead of block [incubator-uniffle]
Posted by "jerqi (via GitHub)" <gi...@apache.org>.
jerqi commented on code in PR #1272:
URL: https://github.com/apache/incubator-uniffle/pull/1272#discussion_r1375435026
##########
common/src/main/java/org/apache/uniffle/common/metrics/RPCMetrics.java:
##########
@@ -18,27 +18,44 @@
package org.apache.uniffle.common.metrics;
import java.util.Map;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
import com.google.common.collect.Maps;
+import com.google.common.collect.Queues;
import io.prometheus.client.CollectorRegistry;
import io.prometheus.client.Counter;
import io.prometheus.client.Gauge;
import io.prometheus.client.Summary;
import org.apache.uniffle.common.util.Constants;
import org.apache.uniffle.common.util.JavaUtils;
+import org.apache.uniffle.common.util.ThreadUtils;
public abstract class RPCMetrics {
protected boolean isRegistered = false;
protected Map<String, Counter.Child> counterMap = JavaUtils.newConcurrentMap();
protected Map<String, Gauge.Child> gaugeMap = JavaUtils.newConcurrentMap();
protected Map<String, Summary.Child> transportTimeSummaryMap = JavaUtils.newConcurrentMap();
protected Map<String, Summary.Child> processTimeSummaryMap = JavaUtils.newConcurrentMap();
+ private final ExecutorService summaryObservePool;
protected MetricsManager metricsManager;
protected String tags;
public RPCMetrics(String tags) {
this.tags = tags;
+ BlockingQueue<Runnable> waitQueue = Queues.newLinkedBlockingQueue(1000);
Review Comment:
Could we avoid using constant here?
--
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: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
Re: [PR] [#1271] improvement(server): change transportTime and processTime summary to Thread Pool Instead of block [incubator-uniffle]
Posted by "jerqi (via GitHub)" <gi...@apache.org>.
jerqi commented on code in PR #1272:
URL: https://github.com/apache/incubator-uniffle/pull/1272#discussion_r1376031842
##########
common/src/main/java/org/apache/uniffle/common/metrics/RPCMetrics.java:
##########
@@ -18,27 +18,45 @@
package org.apache.uniffle.common.metrics;
import java.util.Map;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
import com.google.common.collect.Maps;
+import com.google.common.collect.Queues;
import io.prometheus.client.CollectorRegistry;
import io.prometheus.client.Counter;
import io.prometheus.client.Gauge;
import io.prometheus.client.Summary;
import org.apache.uniffle.common.util.Constants;
import org.apache.uniffle.common.util.JavaUtils;
+import org.apache.uniffle.common.util.ThreadUtils;
public abstract class RPCMetrics {
protected boolean isRegistered = false;
protected Map<String, Counter.Child> counterMap = JavaUtils.newConcurrentMap();
protected Map<String, Gauge.Child> gaugeMap = JavaUtils.newConcurrentMap();
protected Map<String, Summary.Child> transportTimeSummaryMap = JavaUtils.newConcurrentMap();
protected Map<String, Summary.Child> processTimeSummaryMap = JavaUtils.newConcurrentMap();
+ private static final int WAIT_QUEUE_SIZE = 1000;
+ private final ExecutorService summaryObservePool;
protected MetricsManager metricsManager;
protected String tags;
public RPCMetrics(String tags) {
this.tags = tags;
+ BlockingQueue<Runnable> waitQueue = Queues.newLinkedBlockingQueue(WAIT_QUEUE_SIZE);
+ this.summaryObservePool =
+ new ThreadPoolExecutor(
+ 2,
+ 10,
+ 60,
+ TimeUnit.SECONDS,
+ waitQueue,
+ ThreadUtils.getThreadFactory("SummaryObserveThreadPool"),
+ new ThreadPoolExecutor.DiscardPolicy());
Review Comment:
Block queue seems better. Discard policy will produce the incorrect metrics.
--
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: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
Re: [PR] [#1271] improvement(server): change transportTime and processTime summary to Thread Pool Instead of block [incubator-uniffle]
Posted by "lifeSo (via GitHub)" <gi...@apache.org>.
lifeSo commented on code in PR #1272:
URL: https://github.com/apache/incubator-uniffle/pull/1272#discussion_r1375735368
##########
common/src/main/java/org/apache/uniffle/common/metrics/RPCMetrics.java:
##########
@@ -18,27 +18,45 @@
package org.apache.uniffle.common.metrics;
import java.util.Map;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
import com.google.common.collect.Maps;
+import com.google.common.collect.Queues;
import io.prometheus.client.CollectorRegistry;
import io.prometheus.client.Counter;
import io.prometheus.client.Gauge;
import io.prometheus.client.Summary;
import org.apache.uniffle.common.util.Constants;
import org.apache.uniffle.common.util.JavaUtils;
+import org.apache.uniffle.common.util.ThreadUtils;
public abstract class RPCMetrics {
protected boolean isRegistered = false;
protected Map<String, Counter.Child> counterMap = JavaUtils.newConcurrentMap();
protected Map<String, Gauge.Child> gaugeMap = JavaUtils.newConcurrentMap();
protected Map<String, Summary.Child> transportTimeSummaryMap = JavaUtils.newConcurrentMap();
protected Map<String, Summary.Child> processTimeSummaryMap = JavaUtils.newConcurrentMap();
+ private static final int WAIT_QUEUE_SIZE = 1000;
+ private final ExecutorService summaryObservePool;
protected MetricsManager metricsManager;
protected String tags;
public RPCMetrics(String tags) {
this.tags = tags;
+ BlockingQueue<Runnable> waitQueue = Queues.newLinkedBlockingQueue(WAIT_QUEUE_SIZE);
+ this.summaryObservePool =
+ new ThreadPoolExecutor(
+ 2,
Review Comment:
ok, changed it.
--
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: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
Re: [PR] [Improvement]: Change transportTime and processTime summary to Thread Pool Instead of block [incubator-uniffle]
Posted by "codecov-commenter (via GitHub)" <gi...@apache.org>.
codecov-commenter commented on PR #1272:
URL: https://github.com/apache/incubator-uniffle/pull/1272#issuecomment-1783792862
We're currently processing your upload. This comment will be updated when the results are available.
--
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: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
Re: [PR] [#1271] improvement(server): change transportTime and processTime summary to Thread Pool Instead of block [incubator-uniffle]
Posted by "lifeSo (via GitHub)" <gi...@apache.org>.
lifeSo commented on PR #1272:
URL: https://github.com/apache/incubator-uniffle/pull/1272#issuecomment-1784364709
> @lifeSo Could you change your title? We should use `[#xxxx]` or `[MINOR]` as the start of our title.
Ok, I changed it.
--
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: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
Re: [PR] [#1271] improvement(server): change transportTime and processTime summary to Thread Pool Instead of block [incubator-uniffle]
Posted by "jerqi (via GitHub)" <gi...@apache.org>.
jerqi commented on PR #1272:
URL: https://github.com/apache/incubator-uniffle/pull/1272#issuecomment-1784558117
> > > > What will it happen if we don't have this pr?
> > >
> > >
> > > GRPC Thread will block on the metric calc. Here is the jstack from PRD, We find many GRPC Thread blocks. So, if we don't have this pr, the GRPC Thread will block.
> > > 
> >
> >
> > Will it cause many requests timeout?
>
> It will cause the GRPC block, then GRPC request queue come with many request task. Some client will read shuffle data fail
OK, I know.
--
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: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
Re: [PR] [#1271] improvement(server): change transportTime and processTime summary to Thread Pool Instead of block [incubator-uniffle]
Posted by "jerqi (via GitHub)" <gi...@apache.org>.
jerqi commented on code in PR #1272:
URL: https://github.com/apache/incubator-uniffle/pull/1272#discussion_r1375722936
##########
common/src/main/java/org/apache/uniffle/common/metrics/RPCMetrics.java:
##########
@@ -18,27 +18,45 @@
package org.apache.uniffle.common.metrics;
import java.util.Map;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
import com.google.common.collect.Maps;
+import com.google.common.collect.Queues;
import io.prometheus.client.CollectorRegistry;
import io.prometheus.client.Counter;
import io.prometheus.client.Gauge;
import io.prometheus.client.Summary;
import org.apache.uniffle.common.util.Constants;
import org.apache.uniffle.common.util.JavaUtils;
+import org.apache.uniffle.common.util.ThreadUtils;
public abstract class RPCMetrics {
protected boolean isRegistered = false;
protected Map<String, Counter.Child> counterMap = JavaUtils.newConcurrentMap();
protected Map<String, Gauge.Child> gaugeMap = JavaUtils.newConcurrentMap();
protected Map<String, Summary.Child> transportTimeSummaryMap = JavaUtils.newConcurrentMap();
protected Map<String, Summary.Child> processTimeSummaryMap = JavaUtils.newConcurrentMap();
+ private static final int WAIT_QUEUE_SIZE = 1000;
Review Comment:
Could we use config option instead of constant here?
##########
common/src/main/java/org/apache/uniffle/common/metrics/RPCMetrics.java:
##########
@@ -18,27 +18,45 @@
package org.apache.uniffle.common.metrics;
import java.util.Map;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
import com.google.common.collect.Maps;
+import com.google.common.collect.Queues;
import io.prometheus.client.CollectorRegistry;
import io.prometheus.client.Counter;
import io.prometheus.client.Gauge;
import io.prometheus.client.Summary;
import org.apache.uniffle.common.util.Constants;
import org.apache.uniffle.common.util.JavaUtils;
+import org.apache.uniffle.common.util.ThreadUtils;
public abstract class RPCMetrics {
protected boolean isRegistered = false;
protected Map<String, Counter.Child> counterMap = JavaUtils.newConcurrentMap();
protected Map<String, Gauge.Child> gaugeMap = JavaUtils.newConcurrentMap();
protected Map<String, Summary.Child> transportTimeSummaryMap = JavaUtils.newConcurrentMap();
protected Map<String, Summary.Child> processTimeSummaryMap = JavaUtils.newConcurrentMap();
+ private static final int WAIT_QUEUE_SIZE = 1000;
+ private final ExecutorService summaryObservePool;
protected MetricsManager metricsManager;
protected String tags;
public RPCMetrics(String tags) {
this.tags = tags;
+ BlockingQueue<Runnable> waitQueue = Queues.newLinkedBlockingQueue(WAIT_QUEUE_SIZE);
+ this.summaryObservePool =
+ new ThreadPoolExecutor(
+ 2,
Review Comment:
Could we use config option instead of constant here?
--
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: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
Re: [PR] [#1271] improvement(server): change transportTime and processTime summary to Thread Pool Instead of block [incubator-uniffle]
Posted by "lifeSo (via GitHub)" <gi...@apache.org>.
lifeSo commented on code in PR #1272:
URL: https://github.com/apache/incubator-uniffle/pull/1272#discussion_r1375729378
##########
common/src/main/java/org/apache/uniffle/common/metrics/RPCMetrics.java:
##########
@@ -18,27 +18,45 @@
package org.apache.uniffle.common.metrics;
import java.util.Map;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
import com.google.common.collect.Maps;
+import com.google.common.collect.Queues;
import io.prometheus.client.CollectorRegistry;
import io.prometheus.client.Counter;
import io.prometheus.client.Gauge;
import io.prometheus.client.Summary;
import org.apache.uniffle.common.util.Constants;
import org.apache.uniffle.common.util.JavaUtils;
+import org.apache.uniffle.common.util.ThreadUtils;
public abstract class RPCMetrics {
protected boolean isRegistered = false;
protected Map<String, Counter.Child> counterMap = JavaUtils.newConcurrentMap();
protected Map<String, Gauge.Child> gaugeMap = JavaUtils.newConcurrentMap();
protected Map<String, Summary.Child> transportTimeSummaryMap = JavaUtils.newConcurrentMap();
protected Map<String, Summary.Child> processTimeSummaryMap = JavaUtils.newConcurrentMap();
+ private static final int WAIT_QUEUE_SIZE = 1000;
+ private final ExecutorService summaryObservePool;
protected MetricsManager metricsManager;
protected String tags;
public RPCMetrics(String tags) {
this.tags = tags;
+ BlockingQueue<Runnable> waitQueue = Queues.newLinkedBlockingQueue(WAIT_QUEUE_SIZE);
+ this.summaryObservePool =
+ new ThreadPoolExecutor(
+ 2,
+ 10,
+ 60,
+ TimeUnit.SECONDS,
+ waitQueue,
+ ThreadUtils.getThreadFactory("SummaryObserveThreadPool"),
+ new ThreadPoolExecutor.DiscardPolicy());
Review Comment:
We consider use block queue, but it will block many metric.
--
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: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
Re: [PR] [#1271] improvement(server): change transportTime and processTime summary to Thread Pool Instead of block [incubator-uniffle]
Posted by "lifeSo (via GitHub)" <gi...@apache.org>.
lifeSo commented on code in PR #1272:
URL: https://github.com/apache/incubator-uniffle/pull/1272#discussion_r1375735368
##########
common/src/main/java/org/apache/uniffle/common/metrics/RPCMetrics.java:
##########
@@ -18,27 +18,45 @@
package org.apache.uniffle.common.metrics;
import java.util.Map;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
import com.google.common.collect.Maps;
+import com.google.common.collect.Queues;
import io.prometheus.client.CollectorRegistry;
import io.prometheus.client.Counter;
import io.prometheus.client.Gauge;
import io.prometheus.client.Summary;
import org.apache.uniffle.common.util.Constants;
import org.apache.uniffle.common.util.JavaUtils;
+import org.apache.uniffle.common.util.ThreadUtils;
public abstract class RPCMetrics {
protected boolean isRegistered = false;
protected Map<String, Counter.Child> counterMap = JavaUtils.newConcurrentMap();
protected Map<String, Gauge.Child> gaugeMap = JavaUtils.newConcurrentMap();
protected Map<String, Summary.Child> transportTimeSummaryMap = JavaUtils.newConcurrentMap();
protected Map<String, Summary.Child> processTimeSummaryMap = JavaUtils.newConcurrentMap();
+ private static final int WAIT_QUEUE_SIZE = 1000;
+ private final ExecutorService summaryObservePool;
protected MetricsManager metricsManager;
protected String tags;
public RPCMetrics(String tags) {
this.tags = tags;
+ BlockingQueue<Runnable> waitQueue = Queues.newLinkedBlockingQueue(WAIT_QUEUE_SIZE);
+ this.summaryObservePool =
+ new ThreadPoolExecutor(
+ 2,
Review Comment:
ok, changed it.
--
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: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
Re: [PR] [#1271] improvement(server): change transportTime and processTime summary to Thread Pool Instead of block [incubator-uniffle]
Posted by "lifeSo (via GitHub)" <gi...@apache.org>.
lifeSo commented on code in PR #1272:
URL: https://github.com/apache/incubator-uniffle/pull/1272#discussion_r1376999215
##########
common/src/main/java/org/apache/uniffle/common/metrics/RPCMetrics.java:
##########
@@ -18,27 +18,45 @@
package org.apache.uniffle.common.metrics;
import java.util.Map;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
import com.google.common.collect.Maps;
+import com.google.common.collect.Queues;
import io.prometheus.client.CollectorRegistry;
import io.prometheus.client.Counter;
import io.prometheus.client.Gauge;
import io.prometheus.client.Summary;
import org.apache.uniffle.common.util.Constants;
import org.apache.uniffle.common.util.JavaUtils;
+import org.apache.uniffle.common.util.ThreadUtils;
public abstract class RPCMetrics {
protected boolean isRegistered = false;
protected Map<String, Counter.Child> counterMap = JavaUtils.newConcurrentMap();
protected Map<String, Gauge.Child> gaugeMap = JavaUtils.newConcurrentMap();
protected Map<String, Summary.Child> transportTimeSummaryMap = JavaUtils.newConcurrentMap();
protected Map<String, Summary.Child> processTimeSummaryMap = JavaUtils.newConcurrentMap();
+ private static final int WAIT_QUEUE_SIZE = 1000;
+ private final ExecutorService summaryObservePool;
protected MetricsManager metricsManager;
protected String tags;
public RPCMetrics(String tags) {
this.tags = tags;
+ BlockingQueue<Runnable> waitQueue = Queues.newLinkedBlockingQueue(WAIT_QUEUE_SIZE);
+ this.summaryObservePool =
+ new ThreadPoolExecutor(
+ 2,
+ 10,
+ 60,
+ TimeUnit.SECONDS,
+ waitQueue,
+ ThreadUtils.getThreadFactory("SummaryObserveThreadPool"),
+ new ThreadPoolExecutor.DiscardPolicy());
Review Comment:
Ok，I changed to block queue, and it is equal to modify metric calc to async by thread pool
##########
common/src/main/java/org/apache/uniffle/common/metrics/RPCMetrics.java:
##########
@@ -18,27 +18,45 @@
package org.apache.uniffle.common.metrics;
import java.util.Map;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
import com.google.common.collect.Maps;
+import com.google.common.collect.Queues;
import io.prometheus.client.CollectorRegistry;
import io.prometheus.client.Counter;
import io.prometheus.client.Gauge;
import io.prometheus.client.Summary;
import org.apache.uniffle.common.util.Constants;
import org.apache.uniffle.common.util.JavaUtils;
+import org.apache.uniffle.common.util.ThreadUtils;
public abstract class RPCMetrics {
protected boolean isRegistered = false;
protected Map<String, Counter.Child> counterMap = JavaUtils.newConcurrentMap();
protected Map<String, Gauge.Child> gaugeMap = JavaUtils.newConcurrentMap();
protected Map<String, Summary.Child> transportTimeSummaryMap = JavaUtils.newConcurrentMap();
protected Map<String, Summary.Child> processTimeSummaryMap = JavaUtils.newConcurrentMap();
+ private static final int WAIT_QUEUE_SIZE = 1000;
+ private final ExecutorService summaryObservePool;
protected MetricsManager metricsManager;
protected String tags;
public RPCMetrics(String tags) {
this.tags = tags;
+ BlockingQueue<Runnable> waitQueue = Queues.newLinkedBlockingQueue(WAIT_QUEUE_SIZE);
+ this.summaryObservePool =
+ new ThreadPoolExecutor(
+ 2,
+ 10,
+ 60,
+ TimeUnit.SECONDS,
+ waitQueue,
+ ThreadUtils.getThreadFactory("SummaryObserveThreadPool"),
+ new ThreadPoolExecutor.DiscardPolicy());
Review Comment:
Ok，I changed to block queue, and it is equal to modify metric calc to async by thread pool
--
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: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
Re: [PR] [#1271] improvement(server): change transportTime and processTime summary to Thread Pool Instead of block [incubator-uniffle]
Posted by "jerqi (via GitHub)" <gi...@apache.org>.
jerqi commented on code in PR #1272:
URL: https://github.com/apache/incubator-uniffle/pull/1272#discussion_r1376112553
##########
common/src/main/java/org/apache/uniffle/common/metrics/RPCMetrics.java:
##########
@@ -18,27 +18,45 @@
package org.apache.uniffle.common.metrics;
import java.util.Map;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
import com.google.common.collect.Maps;
+import com.google.common.collect.Queues;
import io.prometheus.client.CollectorRegistry;
import io.prometheus.client.Counter;
import io.prometheus.client.Gauge;
import io.prometheus.client.Summary;
import org.apache.uniffle.common.util.Constants;
import org.apache.uniffle.common.util.JavaUtils;
+import org.apache.uniffle.common.util.ThreadUtils;
public abstract class RPCMetrics {
protected boolean isRegistered = false;
protected Map<String, Counter.Child> counterMap = JavaUtils.newConcurrentMap();
protected Map<String, Gauge.Child> gaugeMap = JavaUtils.newConcurrentMap();
protected Map<String, Summary.Child> transportTimeSummaryMap = JavaUtils.newConcurrentMap();
protected Map<String, Summary.Child> processTimeSummaryMap = JavaUtils.newConcurrentMap();
+ private static final int WAIT_QUEUE_SIZE = 1000;
Review Comment:
Could we use config option? If we find that these parameters aren't suitable for production environment, what can we do?
--
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: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org