CpuThreadCountDataSeries implements DataSeries<Long> { @Override public List<SeriesData<Long>> getDataForRange(Range rangeUs) { long minNs = TimeUnit.MICROSECONDS.toNanos((long)rangeUs.getMin()); long maxNs = TimeUnit.MICROSECONDS.toNanos((long)rangeUs.getMax()); GetEventGroupsRequest request = GetEventGroupsRequest.newBuilder() .setStreamId(myStreamId) .setPid(myPid) .setKind(Common.Event.Kind.CPU_THREAD) .setFromTimestamp(minNs) .setToTimestamp(maxNs) .build(); GetEventGroupsResponse response = myClient.getEventGroups(request); TreeMap<Long, Long> timestampToCountMap = new TreeMap<>(); for (Transport.EventGroup group : response.getGroupsList()) { if (group.getEventsCount() > 0) { Common.Event first = group.getEvents(0); Common.Event last = group.getEvents(group.getEventsCount() - 1); timestampToCountMap.compute(first.getTimestamp(), (timestamp, count) -> count == null ? 1 : count + 1); if (last.getIsEnded()) { timestampToCountMap.compute(last.getTimestamp(), (timestamp, count) -> count == null ? -1 : count - 1); } } } List<SeriesData<Long>> data = new ArrayList<>(); long total = 0; for (Map.Entry<Long, Long> entry : timestampToCountMap.entrySet()) { total += entry.getValue(); data.add(new SeriesData<>(TimeUnit.NANOSECONDS.toMicros(entry.getKey()), total)); } if (timestampToCountMap.isEmpty()) { data.add(new SeriesData<>((long)rangeUs.getMin(), total)); data.add(new SeriesData<>((long)rangeUs.getMax(), total)); } else if (data.get(data.size() - 1).x < rangeUs.getMax()) { data.add(new SeriesData<>((long)rangeUs.getMax(), total)); } return data; } CpuThreadCountDataSeries(@NotNull TransportServiceGrpc.TransportServiceBlockingStub client, long streamId, int pid); @Override List<SeriesData<Long>> getDataForRange(Range rangeUs);  }
@Test public void noAliveThreadsInRange() { double rangeMin = TimeUnit.SECONDS.toMicros(20); double rangeMax = TimeUnit.SECONDS.toMicros(25); Range range = new Range(rangeMin, rangeMax); List<SeriesData<Long>> seriesDataList = myDataSeries.getDataForRange(range); assertEquals(2, seriesDataList.size()); SeriesData<Long> seriesData = seriesDataList.get(0); Objects.requireNonNull(seriesData); assertEquals(rangeMin, seriesData.x, 0); assertEquals(0, (long)seriesData.value); seriesData = seriesDataList.get(1); Objects.requireNonNull(seriesData); assertEquals(rangeMax, seriesData.x, 0); assertEquals(0, (long)seriesData.value); }
@Test public void oneAliveThreadInRange() { Range range = new Range(TimeUnit.SECONDS.toMicros(10), TimeUnit.SECONDS.toMicros(11)); List<SeriesData<Long>> seriesDataList = myDataSeries.getDataForRange(range); assertEquals(2, seriesDataList.size()); SeriesData<Long> seriesData = seriesDataList.get(0); Objects.requireNonNull(seriesData); long timestamp = myIsUnifiedPipeline ? TimeUnit.SECONDS.toMicros(8) : TimeUnit.SECONDS.toMicros(6); assertEquals(timestamp, seriesData.x, 0); assertEquals(1, (long)seriesData.value); seriesData = seriesDataList.get(1); Objects.requireNonNull(seriesData); if (myIsUnifiedPipeline) { assertEquals(TimeUnit.SECONDS.toMicros(11), seriesData.x, 0); assertEquals(1, (long)seriesData.value); } else { assertEquals(TimeUnit.SECONDS.toMicros(15), seriesData.x, 0); assertEquals(0, (long)seriesData.value); } }
@Test public void multipleAliveThreadInRange() { Range range = new Range(TimeUnit.SECONDS.toMicros(6), TimeUnit.SECONDS.toMicros(10)); List<SeriesData<Long>> seriesDataList = myDataSeries.getDataForRange(range); assertEquals(4, seriesDataList.size()); SeriesData<Long> seriesData = seriesDataList.get(0); Objects.requireNonNull(seriesData); assertEquals(TimeUnit.SECONDS.toMicros(1), seriesData.x, 0); assertEquals(1, (long)seriesData.value); seriesData = seriesDataList.get(1); Objects.requireNonNull(seriesData); assertEquals(TimeUnit.SECONDS.toMicros(6), seriesData.x, 0); assertEquals(2, (long)seriesData.value); seriesData = seriesDataList.get(2); Objects.requireNonNull(seriesData); assertEquals(TimeUnit.SECONDS.toMicros(8), seriesData.x, 0); assertEquals(1, (long)seriesData.value); seriesData = seriesDataList.get(3); Objects.requireNonNull(seriesData); if (myIsUnifiedPipeline) { assertEquals(TimeUnit.SECONDS.toMicros(10), seriesData.x, 0); assertEquals(1, (long)seriesData.value); } else { assertEquals(TimeUnit.SECONDS.toMicros(15), seriesData.x, 0); assertEquals(0, (long)seriesData.value); } }
@Test public void threadDiesBeforeRangeMax() { Range range = new Range(TimeUnit.SECONDS.toMicros(10), TimeUnit.SECONDS.toMicros(20)); List<SeriesData<Long>> seriesDataList = myDataSeries.getDataForRange(range); assertEquals(3, seriesDataList.size()); SeriesData<Long> seriesData = seriesDataList.get(0); Objects.requireNonNull(seriesData); long timestamp = myIsUnifiedPipeline ? TimeUnit.SECONDS.toMicros(8) : TimeUnit.SECONDS.toMicros(6); assertEquals(timestamp, seriesData.x, 0); assertEquals(1, (long)seriesData.value); seriesData = seriesDataList.get(1); Objects.requireNonNull(seriesData); assertEquals(TimeUnit.SECONDS.toMicros(15), seriesData.x, 0); assertEquals(0, (long)seriesData.value); seriesData = seriesDataList.get(2); Objects.requireNonNull(seriesData); assertEquals(range.getMax(), seriesData.x, 0); assertEquals(0, (long)seriesData.value); }