Java 类it.unimi.dsi.fastutil.ints.IntOpenHashSet 实例源码

/**
 * compare the matrix meta on the master and the matrix meta on ps to find the matrix this parameter server needs to create and delete
 * @param matrixReports parameter server matrix report, include the matrix ids this parameter server hold.
 * @param needCreateMatrixes use to return the matrix partitions this parameter server need to build
 * @param needReleaseMatrixes use to return the matrix ids this parameter server need to remove
 * @param needRecoverParts need recover partitions
 * @param psId parameter server id
*/
public void syncMatrixInfos(List<MatrixReport> matrixReports,
  List<MatrixMeta> needCreateMatrixes, List<Integer> needReleaseMatrixes,
  List<RecoverPartKey> needRecoverParts, ParameterServerId psId) {

  //get matrix ids in the parameter server report
  IntOpenHashSet matrixInPS = new IntOpenHashSet();
  int size = matrixReports.size();
  for (int i = 0; i < size; i++) {
    matrixInPS.add(matrixReports.get(i).matrixId);
  }

  handleMatrixReports(psId, matrixReports);

  Set<RecoverPartKey> parts = getAndRemoveNeedRecoverParts(psId);
  if(parts != null) {
    needRecoverParts.addAll(parts);
  }

  //get the matrices parameter server need to create and delete
  getPSNeedUpdateMatrix(matrixInPS, needCreateMatrixes, needReleaseMatrixes, psId);
  psMatricesUpdate(psId, matrixReports);
}

@Override
public int hashCode() {
    final int prime = 31;
    int result = 1;

    List<IntOpenHashSet> setCluster = this.convertClustersToSets(this.clusters);

    Collections.sort(setCluster, new Comparator<IntSet>() {
        @Override
        public int compare(IntSet o1, IntSet o2) {
            return o1.hashCode() - o2.hashCode();
        }
    });
    result = prime * result + (setCluster.hashCode());
    return result;
}

@Override
public int hashCode() {
    final int prime = 31;
    int result = 1;

    List<IntOpenHashSet> setCluster = this.convertClustersToSets(this.clusters);

    Collections.sort(setCluster, new Comparator<IntSet>() {
        @Override
        public int compare(IntSet o1, IntSet o2) {
            return o1.hashCode() - o2.hashCode();
        }
    });
    result = prime * result + (setCluster.hashCode());
    return result;
}

private Map<String, IntSet> groupAttributesByValue(final IntSet attributes)
    throws AlgorithmExecutionException {

  final Map<String, IntSet> attributesByValue = new HashMap<>();
  for (final int attribute : attributes) {
    final Collection<String> values = getValues(attribute);

    if (configuration.isProcessEmptyColumns() && values.isEmpty()) {
      handleEmptyAttribute(attribute, attributes);
    } else {
      for (final String value : values) {
        attributesByValue.computeIfAbsent(value, k -> new IntOpenHashSet()).add(attribute);
      }
    }
  }
  return attributesByValue;
}

/**
 * Construct a PartitionedRegionQueryEvaluator
 * 
 * @param sys the distributed system
 * @param pr the partitioned region
 * @param query the query
 * @param parameters the parameters for executing the query
 * @param cumulativeResults where to add the results as they come in
 */
public PartitionedRegionQueryEvaluator(InternalDistributedSystem sys, PartitionedRegion pr,
    DefaultQuery query, Object[] parameters, SelectResults cumulativeResults,
    Set<Integer> bucketsToQuery) {
  super(sys, pr.getPRId());
  this.pr = pr;
  this.query = query;
  this.parameters = parameters;
  this.cumulativeResults = cumulativeResults;
  this.bucketsToQuery = bucketsToQuery;
  this.successfulBuckets = new IntOpenHashSet(this.bucketsToQuery.size());
  this.resultsPerMember =
      new ConcurrentHashMap<InternalDistributedMember, Collection<Collection>>();
  this.node2bucketIds = Collections.emptyMap();
  if (query != null && query.isTraced()) {
    prQueryTraceInfoList = new ConcurrentLinkedQueue();
  }
}

/**
 * Return the ground truth in an RDD format, each entity represented with an integer entity id. 
 * @param entityIds1RDD
 * @param entityIds2RDD
 * @param gt
 * @param GT_SEPARATOR
 * @return 
 */
public static JavaPairRDD<Integer,Integer> getGroundTruthIdsFromEntityIds (JavaRDD<String> entityIds1RDD, JavaRDD<String> entityIds2RDD, JavaRDD<String> gt, String GT_SEPARATOR) {
    Object2IntOpenHashMap<String> entityIds1 = readEntityIdsMapping(entityIds1RDD, true);
    Object2IntOpenHashMap<String> entityIds2 = readEntityIdsMapping(entityIds2RDD, false); 

    return gt.mapToPair(line -> {
                line = line.toLowerCase();
                String [] parts = line.split(GT_SEPARATOR);                    
                parts[1] = encodeURIinUTF8(parts[1]);
                return new Tuple2<>(-entityIds2.getOrDefault(parts[1], -1), //negative id first (keep default -1, since -(-1) == 1)
                                    entityIds1.getOrDefault(parts[0], -1)); //positive id second
            })
            .filter(x-> x._1() != 1 && x._2() != -1) //throw away pairs whose elements (one or both) do not appear in the dataset
            //remove pairs violating the clean-clean constraint
            .aggregateByKey(new IntOpenHashSet(), 
                    (x,y) -> {x.add(y); return x;}, 
                    (x,y) -> {x.addAll(y); return x;})
            .filter(x -> x._2().size() == 1) //not more than one match allowed per (negative) entity
            .mapValues(x -> x.iterator().next());

}

private void replayTrain(@Nonnull final ByteBuffer buf) {
    final int itemI = buf.getInt();
    final int knnSize = buf.getInt();

    final Int2ObjectMap<Int2FloatMap> knnItems = new Int2ObjectOpenHashMap<>(1024);
    final IntSet pairItems = new IntOpenHashSet();
    for (int i = 0; i < knnSize; i++) {
        int user = buf.getInt();
        int ruSize = buf.getInt();
        Int2FloatMap ru = new Int2FloatOpenHashMap(ruSize);
        ru.defaultReturnValue(0.f);

        for (int j = 0; j < ruSize; j++) {
            int itemK = buf.getInt();
            pairItems.add(itemK);
            float ruk = buf.getFloat();
            ru.put(itemK, ruk);
        }
        knnItems.put(user, ru);
    }

    for (int itemJ : pairItems) {
        train(itemI, knnItems, itemJ);
    }
}

@Override
public int hashCode() {
    final int prime = 31;
    int result = 1;

    List<IntOpenHashSet> setCluster = convertClustersToSets(clusters);

    Collections.sort(setCluster, new Comparator<IntSet>() {

        @Override
        public int compare(IntSet o1, IntSet o2) {
            return o1.hashCode() - o2.hashCode();
        }
    });
    result = prime * result + (setCluster.hashCode());
    return result;
}

public GraphFunctions(int[][] edges2D, int noNodes) {
 try {
        nv = noNodes;
        nodes = new IntOpenHashSet[nv];

        for(int i = 0;i < nv;i ++) {
            nodes[i] = new IntOpenHashSet();
        }
        for(int i = 0; i < edges2D.length; i++) {
            int v1 = edges2D[i][0];
            int v2 = edges2D[i][1];
            nodes[v1].add(v2);
            nodes[v2].add(v1);
        }
    } catch(Exception e) {
        e.printStackTrace();
    }
}

public void learnNPassShuffled(int n) throws ReflectiveOperationException, IOException {
    pl.info = classifier.shortStats();
    pl.expectedUpdates = graph.numArcs() * 2 * n;
    pl.start();

    for (int pass = 0; pass < n; pass++) {
        LOGGER.info("Starting learning pass #"+(pass+1)+"...");
        int[] nodes = MathArrays.natural(numNodes);
        nodes = IntArrays.shuffle(nodes, rnd);

        for (int node : nodes)
            learnNode(node, new IntOpenHashSet(
                    graph.successorArray(node), 0, graph.outdegree(node)
                    ));

        save(pass+1);

    }

    pl.done();
}

public int[] createTestingSet(int numOfSamples) {
    numOfSamples = Math.min(numOfSamples, numNodes);

    if (verbose) LOGGER.info("Creating test set with "+numOfSamples+" nodes...");
    if (numOfSamples >= (numNodes/2)) {
        final Random rnd = RandomSingleton.get();
        int[] samples = MathArrays.natural(numNodes);
        IntArrays.shuffle(samples, rnd);
        return IntArrays.trim(samples, numOfSamples);
    } else {
        IntSet set = new IntOpenHashSet();
        while (set.size() < numOfSamples) {
            set.add(rnd.nextInt(numNodes));
        }
        int[] r = set.toIntArray();
        return r;
    }
}

private IntSet parseCategories(Document wikiPage) throws IOException {
        String categoryString = IOUtils.toString((Reader) wikiPage.content(CATEGORY_FIELD));
        IntSet categoryIds = new IntOpenHashSet();
        int pipeIndex;

        for (String category : categoryString.split(SEPARATOR_REGEX)) {
            if ((pipeIndex = category.indexOf('|')) > -1)
                category = category.substring(0, pipeIndex);

            category = StringUtils.strip(category);
            if (category.length() > 0)
                categoryIds.add(getCategoryId(category));
        }

        return categoryIds;
}

public void removeIfNotOneRelevantAndOneIrrelevant() {
    IntSet queriesToRemove = new IntOpenHashSet();
    for (int query : queries()) {
        int nEval = query2evaluateddocs.get(query).size();
        int nRelev = query2relevantdocs.get(query).size();
        int nIrrelev = nEval - nRelev;
        if (nRelev < 1 || nIrrelev < 1)
            queriesToRemove.add(query);
    }

    int originalNQueries = queries().size();
    for (int q : queriesToRemove) {
        query2doc2relevance.remove(q);
        query2evaluateddocs.remove(q);
        query2relevantdocs.remove(q);
    }

    LOGGER.info(queriesToRemove.size() + " queries were without "
            + "relevant or irrilevant results and were removed. "
            + "Number of queries went from " + originalNQueries 
            + " to " + queries().size() + ".");
}

public void compute() {
    ProgressLogger pl = new ProgressLogger(LOGGER, "pages");
    pl.expectedUpdates = page2cat.size();
    pl.start("Moving old categories to closest milestones...");
    for (IntSet entry : page2cat.values()) {
        IntSet newCategories = new IntOpenHashSet();
        int milestone;
        for (int cat : entry) {
            milestone = closestMilestones[cat];
            if (milestone != -1)
                newCategories.add(milestone);
        }
        entry.clear();
        entry.addAll(newCategories);
        pl.lightUpdate();
    }
    pl.done();

}

@Override
public Stream<? extends FMInstance> stream() {
    return uidxs.stream()
            .flatMap(uidx -> {
                IntSet uidxIidxs = new IntOpenHashSet();
                prefs.getUidxIidxs(uidx).forEachRemaining(uidxIidxs::add);

                List<FMInstance> instances = new ArrayList<>();

                // adding positive examples
                uidxIidxs
                        .forEach(iidx -> instances.add(getInstance(uidx, iidx, 1.0)));

                // adding negative examples
                rnd.ints(iidxs.size(), 0, iidxs.size()).map(iidxs::getInt)
                        .filter(jidx -> !uidxIidxs.contains(jidx))
                        .distinct()
                        .limit((int) (negativeProp * uidxIidxs.size()))
                        .forEach(jidx -> instances.add(getInstance(uidx, jidx, 0.0)));

                Collections.shuffle(instances);

                return instances.stream();
            });
}

public void setParameters(int sigma, int gamma, int lambda, Taxonomy taxonomy) {
    this.sigma = sigma;
    this.gamma = gamma;
    this.lambda = lambda;
    this.taxonomy = taxonomy;

    scanIndex = new IntOpenHashSet[lambda - 1][lambda - 1];
    rightIndex = new IntOpenHashSet[lambda - 1][lambda - 1];
    for (int i = 0; i < lambda - 1; ++i) {
        for (int j = 0; j < lambda - 1; ++j) {
            scanIndex[i][j] = new IntOpenHashSet();
            rightIndex[i][j] = new IntOpenHashSet();
        }
    }
    clear();
}

/**
 * Baseline implementation. Augments the "standard" list with alternatives.
 * 
 * @param l1
 * @param l2
 * @return
 */
public static <TK,FV> List<RichTranslation<TK,FV>> mergeAndDedup(List<RichTranslation<TK,FV>> standard,
    List<RichTranslation<TK,FV>> alt, int maxAltItems) {

  IntSet hashCodeSet = new IntOpenHashSet(standard.size());
  for (RichTranslation<TK,FV> s : standard) {
    hashCodeSet.add(derivationHashCode(s.getFeaturizable().derivation));
  }

  List<RichTranslation<TK,FV>> returnList = new ArrayList<>(standard);
  for (int i = 0, sz = Math.min(maxAltItems, alt.size()); i < sz; ++i) {
    RichTranslation<TK,FV> t = alt.get(i);
    int hashCode = derivationHashCode(t.getFeaturizable().derivation);
    if (! hashCodeSet.contains(hashCode)) returnList.add(t);
  }
  Collections.sort(returnList);

  return returnList;
}

@Override
public IntOpenHashSet aggregate(Block docIdSetBlock, Block[] block) {
  IntOpenHashSet ret = new IntOpenHashSet();
  int docId = 0;
  BlockDocIdIterator docIdIterator = docIdSetBlock.getBlockDocIdSet().iterator();
  BlockSingleValIterator blockValIterator = (BlockSingleValIterator) block[0].getBlockValueSet().iterator();

  // Assume dictionary is always there for String data type.
  // If data type is String, we shouldn't hit here.
  while ((docId = docIdIterator.next()) != Constants.EOF) {
    if (blockValIterator.skipTo(docId)) {
      ret.add(blockValIterator.nextIntVal());
    }
  }

  return ret;
}

public BlockValSetBlockDocIdIterator(BlockValSet blockValSet, BlockMetadata blockMetadata, int[] dictIds) {
  if (dictIds.length > 0) {
    this.dictIdSet = new IntOpenHashSet(dictIds);
    this.intArray = new int[blockMetadata.getMaxNumberOfMultiValues()];
    Arrays.fill(intArray, 0);
    setStartDocId(blockMetadata.getStartDocId());
    setEndDocId(blockMetadata.getEndDocId());
  } else {
    this.dictIdSet = null;
    this.intArray = new int[0];
    setStartDocId(Constants.EOF);
    setEndDocId(Constants.EOF);
    currentDocId = Constants.EOF;
  }
  valueIterator = (BlockMultiValIterator) blockValSet.iterator();
}

@Test
public void testInsertionTime() {
    int numOfItems = 10000000;

    println("#items_inserted, HLL_time(nano), IntOpenHashSet(nano), time_ratio, estimate, precise, error");
    for (int i = 0; i < numOfItems; i+=numOfItems/17) {
        if (i == 0) {
            continue;
        }
        RandomNumberArray arr = new RandomNumberArray(i, DUPLICATION_PER_ITEM);
        HyperLogLog hllResult = new HyperLogLog(DistinctCountHLLAggregationFunction.DEFAULT_BIT_SIZE);
        IntOpenHashSet set = new IntOpenHashSet(); //HashSet<Integer> set = new HashSet<Integer>();
        long t1 = System.nanoTime();
        arr.offerAllNumberTo(hllResult);
        long t2 = System.nanoTime();
        arr.offerAllNumberTo(set);
        long t3 = System.nanoTime();
        long estimate = hllResult.cardinality();
        long precise = set.size();
        println(i + ", " + "" + (t2 - t1) + ", " + (t3 - t2) + ", " + (t2 - t1 + 0.0) / (t3 - t2 + 0.0) + ", "
                + estimate + ", " + precise + ", " + getErrorString(precise, estimate));
    }
    assertEquals(true, true);
}

@Test
public void testMemoryConsumption() {
    int numOfItems = 10000000;

    println("#items_inserted, HLL_ser_size, openHashSet_ser_size, ser_size_ratio, estimate, precise, error");
    for (int i = 0; i < numOfItems; i+=numOfItems/17) {
        if (i == 0) {
            continue;
        }
        RandomNumberArray arr = new RandomNumberArray(i, DUPLICATION_PER_ITEM);
        HyperLogLog hllResult = new HyperLogLog(DistinctCountHLLAggregationFunction.DEFAULT_BIT_SIZE);
        IntOpenHashSet set = new IntOpenHashSet();
        arr.offerAllNumberTo(hllResult);
        arr.offerAllNumberTo(set);
        int hllSize = getSerializedSize(hllResult);
        int setSize = getSerializedSize(set);
        long estimate = hllResult.cardinality();
        long precise = set.size();
        println(i + ", " + hllSize + ", " + setSize + ", " + (hllSize + 0.0) / (setSize + 0.0) + ", "
                + estimate + ", " + precise + ", " + getErrorString(precise, estimate));
    }
    assertEquals(true, true);
}

@Test
public void testDistinctCountAggregation() {
  AggregationFunction aggregationFunction = new DistinctCountAggregationFunction();
  aggregationFunction.init(_paramsInfo);

  // Test aggregate

  // Test combine
  for (int i = 1; i <= _sizeOfCombineList; ++i) {
    List<Serializable> aggregationResults = getIntOpenHashSets(i);
    List<Serializable> combinedResult = aggregationFunction.combine(aggregationResults, CombineLevel.SEGMENT);
    assertEquals(((IntOpenHashSet) (combinedResult.get(0))).size(), i);
  }

  // Test reduce
  for (int i = 1; i <= _sizeOfCombineList; ++i) {
    List<Serializable> combinedResults = getIntOpenHashSets(i);
    int reduceSize = (Integer) aggregationFunction.reduce(combinedResults);
    assertEquals(reduceSize, i);
  }
}

@Test
public void testInnerSegmentPlanMakerForAggregationNoFilter() {
  BrokerRequest brokerRequest = getAggregationNoFilterBrokerRequest();
  PlanMaker instancePlanMaker = new InstancePlanMakerImplV2();
  PlanNode rootPlanNode = instancePlanMaker.makeInnerSegmentPlan(_indexSegment, brokerRequest);
  rootPlanNode.showTree("");
  MAggregationOperator operator = (MAggregationOperator) rootPlanNode.run();
  IntermediateResultsBlock resultBlock = (IntermediateResultsBlock) operator.nextBlock();
  System.out.println(resultBlock.getAggregationResult().get(0));
  System.out.println(resultBlock.getAggregationResult().get(1));
  System.out.println(resultBlock.getAggregationResult().get(2));
  System.out.println(resultBlock.getAggregationResult().get(3));
  System.out.println(resultBlock.getAggregationResult().get(4));
  System.out.println(resultBlock.getAggregationResult().get(5));
  System.out.println(resultBlock.getAggregationResult().get(6));
  Assert.assertEquals(((Number)resultBlock.getAggregationResult().get(0)).longValue(), 2000001L);
  Assert.assertEquals(resultBlock.getAggregationResult().get(1), 2000001000000.0);
  Assert.assertEquals(resultBlock.getAggregationResult().get(2), 2000000.0);
  Assert.assertEquals(resultBlock.getAggregationResult().get(3), 0.0);
  Assert.assertEquals(Double.parseDouble(resultBlock.getAggregationResult().get(4).toString()), 1000000.0);
  Assert.assertEquals(((IntOpenHashSet) resultBlock.getAggregationResult().get(5)).size(), 10);
  Assert.assertEquals(((IntOpenHashSet) resultBlock.getAggregationResult().get(6)).size(), 100);
}

RealtimeDictionaryBasedRangePredicateEvaluator(RangePredicate rangePredicate, MutableDictionary dictionary) {
  _matchingDictIdSet = new IntOpenHashSet();

  int dictionarySize = dictionary.length();
  if (dictionarySize == 0) {
    return;
  }

  String lowerBoundary = rangePredicate.getLowerBoundary();
  String upperBoundary = rangePredicate.getUpperBoundary();
  boolean includeLowerBoundary = rangePredicate.includeLowerBoundary();
  boolean includeUpperBoundary = rangePredicate.includeUpperBoundary();

  if (lowerBoundary.equals("*")) {
    lowerBoundary = dictionary.getMinVal().toString();
  }
  if (upperBoundary.equals("*")) {
    upperBoundary = dictionary.getMaxVal().toString();
  }

  for (int dictId = 0; dictId < dictionarySize; dictId++) {
    if (dictionary.inRange(lowerBoundary, upperBoundary, dictId, includeLowerBoundary, includeUpperBoundary)) {
      _matchingDictIdSet.add(dictId);
    }
  }
}

/**
 * Helper method to serialize an {@link IntOpenHashSet}.
 */
private static byte[] serializeIntOpenHashSet(IntOpenHashSet intOpenHashSet)
    throws IOException {
  ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream();
  DataOutputStream dataOutputStream = new DataOutputStream(byteArrayOutputStream);

  // Write the size of the set.
  dataOutputStream.writeInt(intOpenHashSet.size());

  IntIterator intIterator = intOpenHashSet.iterator();
  while (intIterator.hasNext()) {
    dataOutputStream.writeInt(intIterator.nextInt());
  }

  return byteArrayOutputStream.toByteArray();
}

/**
 * Test for ser/de of {@link IntOpenHashSet}.
 */
@Test
public void testIntOpenHashSet()
    throws IOException {
  for (int i = 0; i < NUM_ITERATIONS; i++) {
    int size = RANDOM.nextInt(100);
    IntOpenHashSet expected = new IntOpenHashSet(size);
    for (int j = 0; j < size; j++) {
      expected.add(RANDOM.nextInt());
    }

    byte[] bytes = ObjectCustomSerDe.serialize(expected);
    IntOpenHashSet actual = ObjectCustomSerDe.deserialize(bytes, ObjectType.IntOpenHashSet);

    // Use Object comparison instead of Collection comparison because order might change.
    Assert.assertEquals((Object) actual, expected, ERROR_MESSAGE);
  }
}

@Override
public List<Integer> call() throws Exception {
  LogContext.startDataset(datasetKey);
  int counter = 0;
  LOG.debug("Starting usage sync");
  usageKeys = Maps.newHashMap();
  inserts = new IntOpenHashSet();
  List<Integer> neoKeys = Lists.newArrayList();
  for (List<Integer> neoBatch : Iterables.partition(usages, BATCH_SIZE)) {
    if (firstId < 0) {
      firstId = neoBatch.get(0);
    }
    neoKeys.addAll(neoBatch);
    write(neoBatch);
    counter = counter + neoBatch.size();
  }
  LOG.info("Completed batch of {} usages, starting with id {}.", counter, firstId);
  LogContext.endDataset();

  // submit extension sync job for all usages
  ExtensionSync eSync = new ExtensionSync(dao, datasetKey, firstId, usageKeys, inserts);
  dao.reportNewFuture(addTask(eSync));

  return neoKeys;
}

int calcRandomHits(int _size, int _seed) {
  IntSet _cache = new IntOpenHashSet();
  IntList _list = new IntArrayList();
  Random _random = new Random(_seed);
  int _hitCnt = 0;
  for (int v : getTrace()) {
    if(_cache.contains(v)) {
      _hitCnt++;
    } else {
      if (_cache.size() == _size) {
        int cnt = _random.nextInt(_cache.size());
        _cache.remove(_list.get(cnt));
        _list.remove(cnt);
      }
      _cache.add(v);
      _list.add(v);
    }
  }
  return _hitCnt;
}

/**
 * Create a new MatrixClientAdapter.
 */
public MatrixClientAdapter() {
  locks = new ConcurrentHashMap<Integer, ReentrantLock>();
  resultsMap = new ConcurrentHashMap<RowIndex, GetRowsResult>();
  fetchingRowSets = new ConcurrentHashMap<Integer, IntOpenHashSet>();
  matrixToRowSplitSizeCache = new HashMap<Integer, Int2IntOpenHashMap>();
  requestToResponseMap = new ConcurrentHashMap<UserRequest, PartitionResponseCache>();
  workerPool = Executors.newCachedThreadPool();
  stopped = new AtomicBoolean(false);
  syncClockEnable = PSAgentContext.get().syncClockEnable();
}

/**
 * Get rows use pipeline mode.
 *
 * @param result       result cache
 * @param rowIndex     the indexes of rows that need to fetch from ps
 * @param rpcBatchSize how many rows to be fetched in a rpc
 * @param clock        clock value
 * @return result cache
 */
public GetRowsResult getRowsFlow(GetRowsResult result, RowIndex rowIndex, int rpcBatchSize,
  int clock) {
  LOG.debug("get rows request, rowIndex=" + rowIndex);
  if (rpcBatchSize == -1) {
    rpcBatchSize = chooseRpcBatchSize(rowIndex);
  }

  // Filter the rowIds which are fetching now
  ReentrantLock lock = getLock(rowIndex.getMatrixId());
  RowIndex needFetchRows = null;
  try {
    lock.lock();
    resultsMap.put(rowIndex, result);

    if (!fetchingRowSets.containsKey(rowIndex.getMatrixId())) {
      fetchingRowSets.put(rowIndex.getMatrixId(), new IntOpenHashSet());
    }

    if (!matrixToRowSplitSizeCache.containsKey(rowIndex.getMatrixId())) {
      matrixToRowSplitSizeCache.put(rowIndex.getMatrixId(), new Int2IntOpenHashMap());
    }

    needFetchRows = findNewRows(rowIndex);
  } finally {
    lock.unlock();
  }

  // Send the rowIndex to rpc dispatcher and return immediately
  if (needFetchRows.getRowsNumber() > 0) {
    dispatchGetRows(needFetchRows, rpcBatchSize, clock);
  }
  return resultsMap.get(rowIndex);
}

private void notifyAllGetRows(TVector row) {
  if(row == null) {
    return;
  }
  PSAgentContext.get().getMatrixStorageManager().addRow(row.getMatrixId(), row.getRowId(), row);
  ReentrantLock lock = getLock(row.getMatrixId());
  try {
    lock.lock();

    Iterator<Entry<RowIndex, GetRowsResult>> iter = resultsMap.entrySet().iterator();
    Entry<RowIndex, GetRowsResult> resultEntry = null;
    while (iter.hasNext()) {
      resultEntry = iter.next();
      if (resultEntry.getKey().getMatrixId() == row.getMatrixId() && resultEntry.getKey()
        .contains(row.getRowId()) && !resultEntry.getKey().isFilted(row.getRowId())) {
        resultEntry.getKey().filted(row.getRowId());
        resultEntry.getValue().put(row);
      }

      if (resultEntry.getKey().getRowsNumber() == resultEntry.getValue().getRowsNumber()) {
        resultEntry.getKey().clearFilted();
        resultEntry.getValue().fetchOver();
        iter.remove();
      }
    }

    IntOpenHashSet fetchingRowsForMatrix = fetchingRowSets.get(row.getMatrixId());
    if (fetchingRowsForMatrix != null) {
      fetchingRowsForMatrix.remove(row.getRowId());
    }
  } catch (InterruptedException e) {
    LOG.error("Interrupted when notify getrowrequest, exit now ", e);
  } finally {
    lock.unlock();
  }
}

private RowIndex findNewRows(RowIndex rowIndex) {
  IntOpenHashSet need = new IntOpenHashSet();
  IntOpenHashSet fetchingRowIds = fetchingRowSets.get(rowIndex.getMatrixId());

  IntIterator iter = rowIndex.getRowIds().iterator();
  while (iter.hasNext()) {
    int rowId = iter.nextInt();
    if (!fetchingRowIds.contains(rowId)) {
      need.add(rowId);
      fetchingRowIds.add(rowId);
    }
  }

  return new RowIndex(rowIndex.getMatrixId(), need, rowIndex);
}

/**
 * Create a new RowIndex.
 *
 * @param matrixId matrix id
 * @param rowIdSet need fetch row index set
 * @param parent parent RowIndex instance
 */
public RowIndex(int matrixId, IntOpenHashSet rowIdSet, RowIndex parent) {
  this.matrixId = matrixId;
  this.rowIdSet = rowIdSet;
  this.parent = parent;
  this.uuid = UUID.randomUUID();
  this.filtedIdSet = new IntOpenHashSet();
}

public LabeledUpdateIndexBaseTask(TaskContext taskContext, MatrixMeta matrixMeta)
        throws IOException {
  super(taskContext);
  this.matrixMeta = matrixMeta;
  updateIndexEnable = true;
  indexSet = new IntOpenHashSet();
}

protected List<IntOpenHashSet> convertClustersToSets(List<IntArrayList> listCluster) {
    List<IntOpenHashSet> setClusters = new LinkedList<>();
    for (IntArrayList cluster : listCluster) {
        setClusters.add(new IntOpenHashSet(cluster));
    }

    return setClusters;
}

protected List<IntOpenHashSet> convertClustersToSets(List<IntArrayList> listCluster) {
    List<IntOpenHashSet> setClusters = new LinkedList<>();
    for (IntArrayList cluster : listCluster) {
        setClusters.add(new IntOpenHashSet(cluster));
    }

    return setClusters;
}

private IntSet allIds() {
  final IntSet ids = new IntOpenHashSet(attributeIndex.length);
  for (int index = 0; index < attributeIndex.length; ++index) {
    ids.add(index);
  }
  return ids;
}

private Map<String, IntSet> groupAttributesByType() {
  final Map<String, IntSet> attributesByType = new HashMap<>();
  for (final Attribute attribute : attributeIndex) {
    attributesByType
        .computeIfAbsent(attribute.getType(), k -> new IntOpenHashSet())
        .add(attribute.getId());
  }
  return attributesByType;
}

private IntSet[] computeClosures(final Map<String, IntSet> attributesByValue) {
  final IntSet[] closures = new IntSet[attributeIndex.length];
  for (Map.Entry<String, IntSet> entry : attributesByValue.entrySet()) {
    for (int attribute : entry.getValue()) {
      if (closures[attribute] == null) {
        closures[attribute] = new IntOpenHashSet(entry.getValue());
      } else {
        closures[attribute].retainAll(entry.getValue());
      }
    }
  }
  return closures;
}

/**
 * Initialize the original set of recipient <code>Gateway</code>s.
 * 
 * @param originalGatewaysReceivers The original recipient <code>Gateway</code>s.
 */
public void initializeReceipientDSIds(List<Integer> originalGatewaysReceivers) {
  this.receipientDSIds = new IntOpenHashSet(2);
  for (Integer id : originalGatewaysReceivers) {
    this.receipientDSIds.add(id);
  }
}