com.google.common.collect.TreeMultimap#get ( )源码实例Demo

static Path writeIFile(int keysPerFile, int repeatCount) throws
    IOException {
  TreeMultimap<Integer, Long> dataSet = createDataForIFile(keysPerFile, repeatCount);
  LOG.info("DataSet size : " + dataSet.size());
  Path path = new Path(workDir + "/src", "data_" + System.nanoTime() + ".out");
  FSDataOutputStream out = localFs.create(path);
  //create IFile with RLE
  IFile.Writer writer = new IFile.Writer(defaultConf, out, IntWritable.class
      , LongWritable.class, null, null, null, true);

  for (Integer key : dataSet.keySet()) {
    for (Long value : dataSet.get(key)) {
      writer.append(new IntWritable(key), new LongWritable(value));
      verificationDataSet.put(key, value);
    }
  }
  writer.close();
  out.close();
  return path;
}
 

@Override
public @NonNull TmfModelResponse<@NonNull TmfTreeModel<@NonNull TimeGraphEntryModel>> fetchTree(@NonNull Map<@NonNull String, @NonNull Object> fetchParameters, @Nullable IProgressMonitor monitor) {
    if (fLastEnd == Long.MAX_VALUE) {
        return new TmfModelResponse<>(new TmfTreeModel<>(Collections.emptyList(), filter(Objects.requireNonNull(fTraceEntry), fTidToEntry, fetchParameters)), ITmfResponse.Status.COMPLETED, CommonStatusMessage.COMPLETED);
    }

    fModule.waitForInitialization();
    ITmfStateSystem ss = fModule.getStateSystem();
    if (ss == null) {
        return new TmfModelResponse<>(null, ITmfResponse.Status.FAILED, CommonStatusMessage.ANALYSIS_INITIALIZATION_FAILED);
    }

    /*
     * As we are caching the intermediate result, we only want a single thread to
     * update them.
     */
    synchronized (fBuildMap) {
        boolean complete = ss.waitUntilBuilt(0);
        @NonNull List<@NonNull TimeGraphEntryModel> list = Collections.emptyList();
        /* Don't query empty state system */
        if (ss.getNbAttributes() > 0 && ss.getStartTime() != Long.MIN_VALUE) {
            long end = ss.getCurrentEndTime();
            fLastEnd = Long.max(fLastEnd, ss.getStartTime());

            TreeMultimap<Integer, ITmfStateInterval> threadData = TreeMultimap.create(Comparator.naturalOrder(),
                    Comparator.comparing(ITmfStateInterval::getStartTime));

            /*
             * Create a List with the threads' PPID and EXEC_NAME quarks for the 2D query .
             */
            List<Integer> quarks = new ArrayList<>(ss.getQuarks(Attributes.THREADS, WILDCARD, Attributes.EXEC_NAME));
            quarks.addAll(ss.getQuarks(Attributes.THREADS, WILDCARD, Attributes.PPID));
            quarks.addAll(ss.getQuarks(Attributes.THREADS, WILDCARD, Attributes.PID));
            try {
                for (ITmfStateInterval interval : ss.query2D(quarks, Long.min(fLastEnd, end), end)) {
                    if (monitor != null && monitor.isCanceled()) {
                        return new TmfModelResponse<>(null, ITmfResponse.Status.CANCELLED, CommonStatusMessage.TASK_CANCELLED);
                    }
                    threadData.put(interval.getAttribute(), interval);
                }
            } catch (TimeRangeException | StateSystemDisposedException e) {
                return new TmfModelResponse<>(null, ITmfResponse.Status.FAILED, e.getClass().getName() + ':' + String.valueOf(e.getMessage()));
            }

            // update the trace Entry.
            TimeGraphEntryModel traceEntry = new TimeGraphEntryModel(fTraceId, -1, getTrace().getName(), ss.getStartTime(), end);
            fTraceEntry = traceEntry;

            for (Integer threadQuark : ss.getQuarks(Attributes.THREADS, WILDCARD)) {
                String threadAttributeName = ss.getAttributeName(threadQuark);
                Pair<Integer, Integer> entryKey = Attributes.parseThreadAttributeName(threadAttributeName);
                int threadId = entryKey.getFirst();
                if (threadId < 0) {
                    // ignore the 'unknown' (-1) thread
                    continue;
                }

                int execNameQuark = ss.optQuarkRelative(threadQuark, Attributes.EXEC_NAME);
                int ppidQuark = ss.optQuarkRelative(threadQuark, Attributes.PPID);
                int pidQuark = ss.optQuarkRelative(threadQuark, Attributes.PID);
                NavigableSet<ITmfStateInterval> ppidIntervals = threadData.get(ppidQuark);
                NavigableSet<ITmfStateInterval> pidIntervals = threadData.get(pidQuark);
                for (ITmfStateInterval execNameInterval : threadData.get(execNameQuark)) {
                    if (monitor != null && monitor.isCanceled()) {
                        return new TmfModelResponse<>(null, ITmfResponse.Status.CANCELLED, CommonStatusMessage.TASK_CANCELLED);
                    }
                    updateEntry(threadQuark, entryKey, ppidIntervals, execNameInterval, pidIntervals);
                }
            }

            fLastEnd = end;

            list = filter(traceEntry, fTidToEntry, fetchParameters);
        }

        for (TimeGraphEntryModel model : list) {
            fEntryMetadata.put(model.getId(), model.getMetadata());
        }

        if (complete) {
            fBuildMap.clear();
            fLastEnd = Long.MAX_VALUE;
            return new TmfModelResponse<>(new TmfTreeModel<>(Collections.emptyList(), list), ITmfResponse.Status.COMPLETED, CommonStatusMessage.COMPLETED);
        }

        return new TmfModelResponse<>(new TmfTreeModel<>(Collections.emptyList(), list), ITmfResponse.Status.RUNNING, CommonStatusMessage.RUNNING);
    }
}
 

@Override
public @NonNull TmfModelResponse<@NonNull TimeGraphModel> fetchRowModel(@NonNull Map<@NonNull String, @NonNull Object> fetchParameters, IProgressMonitor monitor) {
    ITmfStateSystem ss = fModule.getStateSystem();
    if (ss == null) {
        return new TmfModelResponse<>(null, ITmfResponse.Status.FAILED, CommonStatusMessage.ANALYSIS_INITIALIZATION_FAILED);
    }

    TreeMultimap<Integer, ITmfStateInterval> intervals = TreeMultimap.create(Comparator.naturalOrder(),
            Comparator.comparing(ITmfStateInterval::getStartTime));
    SelectionTimeQueryFilter filter = FetchParametersUtils.createSelectionTimeQuery(fetchParameters);
    Map<Long, Integer> selectedIdsToQuarks = getSelectedIdsToQuarks(filter);
    Collection<Integer> stateAndSyscallQuarks = addSyscall(selectedIdsToQuarks.values(), ss);
    Collection<Long> times = getTimes(ss, filter);
    try {
        /* Do the actual query */
        for (ITmfStateInterval interval : ss.query2D(stateAndSyscallQuarks, times)) {
            if (monitor != null && monitor.isCanceled()) {
                return new TmfModelResponse<>(null, ITmfResponse.Status.CANCELLED, CommonStatusMessage.TASK_CANCELLED);
            }
            intervals.put(interval.getAttribute(), interval);
        }
    } catch (TimeRangeException | StateSystemDisposedException e) {
        return new TmfModelResponse<>(null, ITmfResponse.Status.FAILED, String.valueOf(e.getMessage()));
    }

    Map<@NonNull Integer, @NonNull Predicate< @NonNull Multimap<@NonNull String, @NonNull Object>>> predicates = new HashMap<>();
    Multimap<@NonNull Integer, @NonNull String> regexesMap = DataProviderParameterUtils.extractRegexFilter(fetchParameters);
    if (regexesMap != null) {
        predicates.putAll(computeRegexPredicate(regexesMap));
    }

    @NonNull List<@NonNull ITimeGraphRowModel> rows = new ArrayList<>();
    for (Entry<Long, Integer> entry : selectedIdsToQuarks.entrySet()) {
        int quark = entry.getValue();
        NavigableSet<ITmfStateInterval> states = intervals.get(quark);
        NavigableSet<ITmfStateInterval> syscalls = intervals.get(ss.optQuarkRelative(quark, Attributes.SYSTEM_CALL));

        if (monitor != null && monitor.isCanceled()) {
            return new TmfModelResponse<>(null, ITmfResponse.Status.CANCELLED, CommonStatusMessage.TASK_CANCELLED);
        }
        List<ITimeGraphState> eventList = new ArrayList<>();
        states.forEach(i -> {
            ITimeGraphState timegraphState = createTimeGraphState(i, syscalls);
            Long key = Objects.requireNonNull(entry.getKey());
            applyFilterAndAddState(eventList, timegraphState, key, predicates, monitor);
        });
        rows.add(new TimeGraphRowModel(entry.getKey(), eventList));
    }
    return new TmfModelResponse<>(new TimeGraphModel(rows), ITmfResponse.Status.COMPLETED, CommonStatusMessage.COMPLETED);
}