@Test public void computeTrueAnswers() throws Exception { File configFile = File.createTempFile("test-workload", "toml"); configFile.deleteOnExit(); try (BufferedWriter writer = Files.newBufferedWriter(Paths.get(configFile.getAbsolutePath()))) { writer.write("directory = \"/tmp\"\n" + "[data]\n" + "tstart = 0\n" + "tend = 10_000\n" + "stream-generator = \"RandomStreamGenerator\"\n" + "interarrivals = {distribution = \"FixedDistribution\", value = 1}\n" + "values = {distribution = \"FixedDistribution\", value = 1}\n" + "[workload]\n" + "enable-parallelism = true\n" ); } Configuration conf = new Configuration(configFile); assert conf.getTstart() == 0; long T = conf.getTend(); int Q = 1000; Workload workload = new Workload(); List<Workload.Query> queries = new ArrayList<>(); workload.put("", queries); Random random = new Random(0); for (int q = 0; q < Q; ++q) { long a = Math.floorMod(random.nextLong(), T), b = Math.floorMod(random.nextLong(), T); long l = Math.min(a, b), r = Math.max(a, b); queries.add(new Workload.Query(Workload.Query.Type.COUNT, l, r, 0, null)); } LRMSPopulateWorkload.computeTrueAnswers(conf, workload); for (Workload.Query q : queries) { assertEquals(q.r - q.l + 1, q.trueAnswer.get()); } }
static void computeTrueAnswers(Configuration conf, Workload workload) throws Exception { long T0 = conf.getTstart(), T1 = conf.getTend(); ArrayList<LongRange> intervals = new ArrayList<>(); ArrayList<Query> queries = new ArrayList<>(); for (List<Query> classQueries: workload.values()) { for (Query q: classQueries) { queries.add(q); intervals.add(new LongRange(q.l + ":" + q.r, q.l, true, q.r, true)); } } int Q = queries.size(); Builder builder = new Builder(intervals.toArray(new LongRange[Q]), T0, T1); LongRangeMultiSet lrms = builder.getMultiSet(false, true); if (!conf.isWorkloadParallelismEnabled()) { int[] matchedIndexes = new int[Q]; long[] N = {0}; conf.getStreamGenerator().generate(T0, T1, op -> { if (op.type != StreamGenerator.Operation.Type.APPEND) { return; } if (++N[0] % 1_000_000 == 0) logger.info("t = {}", op.timestamp); processDataPoint(queries, lrms, matchedIndexes, op.timestamp, op.value); }); } else { int nThreads = ForkJoinPool.getCommonPoolParallelism(); assert nThreads > 0; logger.info("# threads = {}", nThreads); long N; { MutableLong nvals = new MutableLong(0L); try (StreamGenerator streamGenerator = conf.getStreamGenerator()) { streamGenerator.generate(T0, T1, op -> { if (op.type == StreamGenerator.Operation.Type.APPEND) { nvals.increment(); } }); } N = nvals.toLong(); } IntStream.range(0, nThreads).parallel().forEach(threadNum -> { long Nleft = threadNum * (N / nThreads); long Nright = (threadNum != nThreads - 1) ? Nleft + (N / nThreads) : N; logger.info("Thread {}: [{}, {})", threadNum, Nleft, Nright); try (StreamGenerator streamGenerator = conf.getStreamGenerator()) { MutableLong Ncurr = new MutableLong(0L); int[] matchedIndexes = new int[Q]; streamGenerator.generate(T0, T1, op -> { if (op.type != StreamGenerator.Operation.Type.APPEND) { return; } if (Nleft <= Ncurr.toLong() && Ncurr.toLong() < Nright) { if (Ncurr.toLong() % 1_000_000 == 0) logger.info("t = {}", op.timestamp); processDataPoint(queries, lrms, matchedIndexes, op.timestamp, op.value); } else if (Ncurr.toLong() >= Nright) { return; } Ncurr.increment(); }); } catch (Exception e) { throw new RuntimeException(e); } logger.info("Thread {} done", threadNum); }); } }
LRMSPopulateWorkload { static void computeTrueAnswers(Configuration conf, Workload workload) throws Exception { long T0 = conf.getTstart(), T1 = conf.getTend(); ArrayList<LongRange> intervals = new ArrayList<>(); ArrayList<Query> queries = new ArrayList<>(); for (List<Query> classQueries: workload.values()) { for (Query q: classQueries) { queries.add(q); intervals.add(new LongRange(q.l + ":" + q.r, q.l, true, q.r, true)); } } int Q = queries.size(); Builder builder = new Builder(intervals.toArray(new LongRange[Q]), T0, T1); LongRangeMultiSet lrms = builder.getMultiSet(false, true); if (!conf.isWorkloadParallelismEnabled()) { int[] matchedIndexes = new int[Q]; long[] N = {0}; conf.getStreamGenerator().generate(T0, T1, op -> { if (op.type != StreamGenerator.Operation.Type.APPEND) { return; } if (++N[0] % 1_000_000 == 0) logger.info("t = {}", op.timestamp); processDataPoint(queries, lrms, matchedIndexes, op.timestamp, op.value); }); } else { int nThreads = ForkJoinPool.getCommonPoolParallelism(); assert nThreads > 0; logger.info("# threads = {}", nThreads); long N; { MutableLong nvals = new MutableLong(0L); try (StreamGenerator streamGenerator = conf.getStreamGenerator()) { streamGenerator.generate(T0, T1, op -> { if (op.type == StreamGenerator.Operation.Type.APPEND) { nvals.increment(); } }); } N = nvals.toLong(); } IntStream.range(0, nThreads).parallel().forEach(threadNum -> { long Nleft = threadNum * (N / nThreads); long Nright = (threadNum != nThreads - 1) ? Nleft + (N / nThreads) : N; logger.info("Thread {}: [{}, {})", threadNum, Nleft, Nright); try (StreamGenerator streamGenerator = conf.getStreamGenerator()) { MutableLong Ncurr = new MutableLong(0L); int[] matchedIndexes = new int[Q]; streamGenerator.generate(T0, T1, op -> { if (op.type != StreamGenerator.Operation.Type.APPEND) { return; } if (Nleft <= Ncurr.toLong() && Ncurr.toLong() < Nright) { if (Ncurr.toLong() % 1_000_000 == 0) logger.info("t = {}", op.timestamp); processDataPoint(queries, lrms, matchedIndexes, op.timestamp, op.value); } else if (Ncurr.toLong() >= Nright) { return; } Ncurr.increment(); }); } catch (Exception e) { throw new RuntimeException(e); } logger.info("Thread {} done", threadNum); }); } } }
LRMSPopulateWorkload { static void computeTrueAnswers(Configuration conf, Workload workload) throws Exception { long T0 = conf.getTstart(), T1 = conf.getTend(); ArrayList<LongRange> intervals = new ArrayList<>(); ArrayList<Query> queries = new ArrayList<>(); for (List<Query> classQueries: workload.values()) { for (Query q: classQueries) { queries.add(q); intervals.add(new LongRange(q.l + ":" + q.r, q.l, true, q.r, true)); } } int Q = queries.size(); Builder builder = new Builder(intervals.toArray(new LongRange[Q]), T0, T1); LongRangeMultiSet lrms = builder.getMultiSet(false, true); if (!conf.isWorkloadParallelismEnabled()) { int[] matchedIndexes = new int[Q]; long[] N = {0}; conf.getStreamGenerator().generate(T0, T1, op -> { if (op.type != StreamGenerator.Operation.Type.APPEND) { return; } if (++N[0] % 1_000_000 == 0) logger.info("t = {}", op.timestamp); processDataPoint(queries, lrms, matchedIndexes, op.timestamp, op.value); }); } else { int nThreads = ForkJoinPool.getCommonPoolParallelism(); assert nThreads > 0; logger.info("# threads = {}", nThreads); long N; { MutableLong nvals = new MutableLong(0L); try (StreamGenerator streamGenerator = conf.getStreamGenerator()) { streamGenerator.generate(T0, T1, op -> { if (op.type == StreamGenerator.Operation.Type.APPEND) { nvals.increment(); } }); } N = nvals.toLong(); } IntStream.range(0, nThreads).parallel().forEach(threadNum -> { long Nleft = threadNum * (N / nThreads); long Nright = (threadNum != nThreads - 1) ? Nleft + (N / nThreads) : N; logger.info("Thread {}: [{}, {})", threadNum, Nleft, Nright); try (StreamGenerator streamGenerator = conf.getStreamGenerator()) { MutableLong Ncurr = new MutableLong(0L); int[] matchedIndexes = new int[Q]; streamGenerator.generate(T0, T1, op -> { if (op.type != StreamGenerator.Operation.Type.APPEND) { return; } if (Nleft <= Ncurr.toLong() && Ncurr.toLong() < Nright) { if (Ncurr.toLong() % 1_000_000 == 0) logger.info("t = {}", op.timestamp); processDataPoint(queries, lrms, matchedIndexes, op.timestamp, op.value); } else if (Ncurr.toLong() >= Nright) { return; } Ncurr.increment(); }); } catch (Exception e) { throw new RuntimeException(e); } logger.info("Thread {} done", threadNum); }); } }  }
LRMSPopulateWorkload { static void computeTrueAnswers(Configuration conf, Workload workload) throws Exception { long T0 = conf.getTstart(), T1 = conf.getTend(); ArrayList<LongRange> intervals = new ArrayList<>(); ArrayList<Query> queries = new ArrayList<>(); for (List<Query> classQueries: workload.values()) { for (Query q: classQueries) { queries.add(q); intervals.add(new LongRange(q.l + ":" + q.r, q.l, true, q.r, true)); } } int Q = queries.size(); Builder builder = new Builder(intervals.toArray(new LongRange[Q]), T0, T1); LongRangeMultiSet lrms = builder.getMultiSet(false, true); if (!conf.isWorkloadParallelismEnabled()) { int[] matchedIndexes = new int[Q]; long[] N = {0}; conf.getStreamGenerator().generate(T0, T1, op -> { if (op.type != StreamGenerator.Operation.Type.APPEND) { return; } if (++N[0] % 1_000_000 == 0) logger.info("t = {}", op.timestamp); processDataPoint(queries, lrms, matchedIndexes, op.timestamp, op.value); }); } else { int nThreads = ForkJoinPool.getCommonPoolParallelism(); assert nThreads > 0; logger.info("# threads = {}", nThreads); long N; { MutableLong nvals = new MutableLong(0L); try (StreamGenerator streamGenerator = conf.getStreamGenerator()) { streamGenerator.generate(T0, T1, op -> { if (op.type == StreamGenerator.Operation.Type.APPEND) { nvals.increment(); } }); } N = nvals.toLong(); } IntStream.range(0, nThreads).parallel().forEach(threadNum -> { long Nleft = threadNum * (N / nThreads); long Nright = (threadNum != nThreads - 1) ? Nleft + (N / nThreads) : N; logger.info("Thread {}: [{}, {})", threadNum, Nleft, Nright); try (StreamGenerator streamGenerator = conf.getStreamGenerator()) { MutableLong Ncurr = new MutableLong(0L); int[] matchedIndexes = new int[Q]; streamGenerator.generate(T0, T1, op -> { if (op.type != StreamGenerator.Operation.Type.APPEND) { return; } if (Nleft <= Ncurr.toLong() && Ncurr.toLong() < Nright) { if (Ncurr.toLong() % 1_000_000 == 0) logger.info("t = {}", op.timestamp); processDataPoint(queries, lrms, matchedIndexes, op.timestamp, op.value); } else if (Ncurr.toLong() >= Nright) { return; } Ncurr.increment(); }); } catch (Exception e) { throw new RuntimeException(e); } logger.info("Thread {} done", threadNum); }); } }  static void main(String[] args); }
LRMSPopulateWorkload { static void computeTrueAnswers(Configuration conf, Workload workload) throws Exception { long T0 = conf.getTstart(), T1 = conf.getTend(); ArrayList<LongRange> intervals = new ArrayList<>(); ArrayList<Query> queries = new ArrayList<>(); for (List<Query> classQueries: workload.values()) { for (Query q: classQueries) { queries.add(q); intervals.add(new LongRange(q.l + ":" + q.r, q.l, true, q.r, true)); } } int Q = queries.size(); Builder builder = new Builder(intervals.toArray(new LongRange[Q]), T0, T1); LongRangeMultiSet lrms = builder.getMultiSet(false, true); if (!conf.isWorkloadParallelismEnabled()) { int[] matchedIndexes = new int[Q]; long[] N = {0}; conf.getStreamGenerator().generate(T0, T1, op -> { if (op.type != StreamGenerator.Operation.Type.APPEND) { return; } if (++N[0] % 1_000_000 == 0) logger.info("t = {}", op.timestamp); processDataPoint(queries, lrms, matchedIndexes, op.timestamp, op.value); }); } else { int nThreads = ForkJoinPool.getCommonPoolParallelism(); assert nThreads > 0; logger.info("# threads = {}", nThreads); long N; { MutableLong nvals = new MutableLong(0L); try (StreamGenerator streamGenerator = conf.getStreamGenerator()) { streamGenerator.generate(T0, T1, op -> { if (op.type == StreamGenerator.Operation.Type.APPEND) { nvals.increment(); } }); } N = nvals.toLong(); } IntStream.range(0, nThreads).parallel().forEach(threadNum -> { long Nleft = threadNum * (N / nThreads); long Nright = (threadNum != nThreads - 1) ? Nleft + (N / nThreads) : N; logger.info("Thread {}: [{}, {})", threadNum, Nleft, Nright); try (StreamGenerator streamGenerator = conf.getStreamGenerator()) { MutableLong Ncurr = new MutableLong(0L); int[] matchedIndexes = new int[Q]; streamGenerator.generate(T0, T1, op -> { if (op.type != StreamGenerator.Operation.Type.APPEND) { return; } if (Nleft <= Ncurr.toLong() && Ncurr.toLong() < Nright) { if (Ncurr.toLong() % 1_000_000 == 0) logger.info("t = {}", op.timestamp); processDataPoint(queries, lrms, matchedIndexes, op.timestamp, op.value); } else if (Ncurr.toLong() >= Nright) { return; } Ncurr.increment(); }); } catch (Exception e) { throw new RuntimeException(e); } logger.info("Thread {} done", threadNum); }); } }  static void main(String[] args);  }