@Test public void testUpdateActiveRatio() throws Exception { ExecutorService service = Executors.newFixedThreadPool(storageGroups.length); CountDownLatch finished = new CountDownLatch(storageGroups.length); for (int i = 0; i < storageGroups.length; i++) { service.submit(new OfferThreads(storageGroups[i], measurementNum[i], finished)); } finished.await(); for (String storageGroup : storageGroups) { ActiveTimeSeriesCounter.getInstance().updateActiveRatio(storageGroup); double sum = 0; for (String s : storageGroups) { sum += ActiveTimeSeriesCounter.getInstance().getActiveRatio(s); } assertEquals(1.0, sum, 0.001); } for (int i = 0; i < storageGroups.length; i++) { double r = ActiveTimeSeriesCounter.getInstance().getActiveRatio(storageGroups[i]); assertEquals(measurementNum[i] / totalSeriesNum, r, 0.001); } }
@Override public void updateActiveRatio(String storageGroup) { lock.writeLock().lock(); try { HyperLogLog log = storageGroupHllMap.get(storageGroup); if (log == null) { LOGGER.warn( "Unknown SG {} is calling updateActiveRatio. It can be ignored the sg is deleted recently", storageGroup); return; } long activeTimeSeriesNum = storageGroupHllMap.get(storageGroup).cardinality(); if (activeTimeSeriesNum != activeTimeSeriesNumMap.get(storageGroup)) { activeTimeSeriesNumMap.put(storageGroup, activeTimeSeriesNum); double totalActiveTsNum = 0; LOGGER.debug("{}: updating active ratio", Thread.currentThread().getName()); for (double number : activeTimeSeriesNumMap.values()) { totalActiveTsNum += number; } for (Map.Entry<String, Long> entry : activeTimeSeriesNumMap.entrySet()) { double activeRatio = 0; if (totalActiveTsNum > 0) { activeRatio = entry.getValue() / totalActiveTsNum; } activeRatioMap.put(entry.getKey(), activeRatio); LOGGER.debug("{}: storage group {} has an active ratio: {}", Thread.currentThread().getName(), entry.getKey(), activeRatio); } } storageGroupHllMap.put(storageGroup, new HyperLogLog(LOG2M)); } catch (Exception e) { LOGGER.error("Update {} active ratio failed", storageGroup, e); } finally { lock.writeLock().unlock(); } }
ActiveTimeSeriesCounter implements IActiveTimeSeriesCounter { @Override public void updateActiveRatio(String storageGroup) { lock.writeLock().lock(); try { HyperLogLog log = storageGroupHllMap.get(storageGroup); if (log == null) { LOGGER.warn( "Unknown SG {} is calling updateActiveRatio. It can be ignored the sg is deleted recently", storageGroup); return; } long activeTimeSeriesNum = storageGroupHllMap.get(storageGroup).cardinality(); if (activeTimeSeriesNum != activeTimeSeriesNumMap.get(storageGroup)) { activeTimeSeriesNumMap.put(storageGroup, activeTimeSeriesNum); double totalActiveTsNum = 0; LOGGER.debug("{}: updating active ratio", Thread.currentThread().getName()); for (double number : activeTimeSeriesNumMap.values()) { totalActiveTsNum += number; } for (Map.Entry<String, Long> entry : activeTimeSeriesNumMap.entrySet()) { double activeRatio = 0; if (totalActiveTsNum > 0) { activeRatio = entry.getValue() / totalActiveTsNum; } activeRatioMap.put(entry.getKey(), activeRatio); LOGGER.debug("{}: storage group {} has an active ratio: {}", Thread.currentThread().getName(), entry.getKey(), activeRatio); } } storageGroupHllMap.put(storageGroup, new HyperLogLog(LOG2M)); } catch (Exception e) { LOGGER.error("Update {} active ratio failed", storageGroup, e); } finally { lock.writeLock().unlock(); } } }
ActiveTimeSeriesCounter implements IActiveTimeSeriesCounter { @Override public void updateActiveRatio(String storageGroup) { lock.writeLock().lock(); try { HyperLogLog log = storageGroupHllMap.get(storageGroup); if (log == null) { LOGGER.warn( "Unknown SG {} is calling updateActiveRatio. It can be ignored the sg is deleted recently", storageGroup); return; } long activeTimeSeriesNum = storageGroupHllMap.get(storageGroup).cardinality(); if (activeTimeSeriesNum != activeTimeSeriesNumMap.get(storageGroup)) { activeTimeSeriesNumMap.put(storageGroup, activeTimeSeriesNum); double totalActiveTsNum = 0; LOGGER.debug("{}: updating active ratio", Thread.currentThread().getName()); for (double number : activeTimeSeriesNumMap.values()) { totalActiveTsNum += number; } for (Map.Entry<String, Long> entry : activeTimeSeriesNumMap.entrySet()) { double activeRatio = 0; if (totalActiveTsNum > 0) { activeRatio = entry.getValue() / totalActiveTsNum; } activeRatioMap.put(entry.getKey(), activeRatio); LOGGER.debug("{}: storage group {} has an active ratio: {}", Thread.currentThread().getName(), entry.getKey(), activeRatio); } } storageGroupHllMap.put(storageGroup, new HyperLogLog(LOG2M)); } catch (Exception e) { LOGGER.error("Update {} active ratio failed", storageGroup, e); } finally { lock.writeLock().unlock(); } }  }
ActiveTimeSeriesCounter implements IActiveTimeSeriesCounter { @Override public void updateActiveRatio(String storageGroup) { lock.writeLock().lock(); try { HyperLogLog log = storageGroupHllMap.get(storageGroup); if (log == null) { LOGGER.warn( "Unknown SG {} is calling updateActiveRatio. It can be ignored the sg is deleted recently", storageGroup); return; } long activeTimeSeriesNum = storageGroupHllMap.get(storageGroup).cardinality(); if (activeTimeSeriesNum != activeTimeSeriesNumMap.get(storageGroup)) { activeTimeSeriesNumMap.put(storageGroup, activeTimeSeriesNum); double totalActiveTsNum = 0; LOGGER.debug("{}: updating active ratio", Thread.currentThread().getName()); for (double number : activeTimeSeriesNumMap.values()) { totalActiveTsNum += number; } for (Map.Entry<String, Long> entry : activeTimeSeriesNumMap.entrySet()) { double activeRatio = 0; if (totalActiveTsNum > 0) { activeRatio = entry.getValue() / totalActiveTsNum; } activeRatioMap.put(entry.getKey(), activeRatio); LOGGER.debug("{}: storage group {} has an active ratio: {}", Thread.currentThread().getName(), entry.getKey(), activeRatio); } } storageGroupHllMap.put(storageGroup, new HyperLogLog(LOG2M)); } catch (Exception e) { LOGGER.error("Update {} active ratio failed", storageGroup, e); } finally { lock.writeLock().unlock(); } }  @Override void init(String storageGroup); @Override void offer(String storageGroup, String device, String measurement); @Override void updateActiveRatio(String storageGroup); @Override double getActiveRatio(String storageGroup); @Override void delete(String storageGroup); static ActiveTimeSeriesCounter getInstance(); static void clear(); }
ActiveTimeSeriesCounter implements IActiveTimeSeriesCounter { @Override public void updateActiveRatio(String storageGroup) { lock.writeLock().lock(); try { HyperLogLog log = storageGroupHllMap.get(storageGroup); if (log == null) { LOGGER.warn( "Unknown SG {} is calling updateActiveRatio. It can be ignored the sg is deleted recently", storageGroup); return; } long activeTimeSeriesNum = storageGroupHllMap.get(storageGroup).cardinality(); if (activeTimeSeriesNum != activeTimeSeriesNumMap.get(storageGroup)) { activeTimeSeriesNumMap.put(storageGroup, activeTimeSeriesNum); double totalActiveTsNum = 0; LOGGER.debug("{}: updating active ratio", Thread.currentThread().getName()); for (double number : activeTimeSeriesNumMap.values()) { totalActiveTsNum += number; } for (Map.Entry<String, Long> entry : activeTimeSeriesNumMap.entrySet()) { double activeRatio = 0; if (totalActiveTsNum > 0) { activeRatio = entry.getValue() / totalActiveTsNum; } activeRatioMap.put(entry.getKey(), activeRatio); LOGGER.debug("{}: storage group {} has an active ratio: {}", Thread.currentThread().getName(), entry.getKey(), activeRatio); } } storageGroupHllMap.put(storageGroup, new HyperLogLog(LOG2M)); } catch (Exception e) { LOGGER.error("Update {} active ratio failed", storageGroup, e); } finally { lock.writeLock().unlock(); } }  @Override void init(String storageGroup); @Override void offer(String storageGroup, String device, String measurement); @Override void updateActiveRatio(String storageGroup); @Override double getActiveRatio(String storageGroup); @Override void delete(String storageGroup); static ActiveTimeSeriesCounter getInstance(); static void clear();  }