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

private static void loadSparseIntPartition(SparseIntModel model, FSDataInputStream input,
    ModelPartitionMeta partMeta) throws IOException {
  int rowNum = input.readInt();
  int rowId = 0;
  int nnz = 0;
  int totalNNZ = 0;
  Int2IntOpenHashMap row = null;

  for (int i = 0; i < rowNum; i++) {
    rowId = input.readInt();
    nnz = input.readInt();
    totalNNZ = (int) (nnz * (model.col) / (partMeta.getEndCol() - partMeta.getStartCol()));
    row = model.getRow(rowId, partMeta.getPartId(), totalNNZ);
    for (int j = 0; j < nnz; j++) {
      row.put(input.readInt(), input.readInt());
    }
  }
}

/**
 * Load dense double model to int->int maps
 *
 * @param modelDir model save directory path
 * @return model data
 */
public static Int2IntOpenHashMap[] loadToIntMaps(String modelDir, Configuration conf)
    throws IOException {
  // Load model meta
  ModelFilesMeta meta = getMeta(modelDir, conf);

  // Check row type
  if (meta.getRowType() != SPARSE_INT) {
    throw new IOException("model row type is not sparse int, you should check it");
  }

  // Load model
  SparseIntModel model = new SparseIntModel(meta.getRow(), meta.getCol());
  loadModel(modelDir, model, meta, conf);

  return model.getModel();
}

private static void convertSparseIntModel(Configuration conf, FSDataOutputStream output,
  String modelInputDir, ModelLineConvert lineConvert) throws IOException {
  Int2IntOpenHashMap[] data = ModelLoader.loadToIntMaps(modelInputDir, conf);
  for(int i = 0; i < data.length; i++) {
    Int2IntOpenHashMap row = data[i];
    data[i] = null;
    if(row == null) {
      continue;
    }

    lineConvert.convertRowIndex(output, i);
    int [] indexes = row.keySet().toIntArray();
    int [] values = row.values().toIntArray();
    row = null;
    Sort.quickSort(indexes, values, 0, indexes.length - 1);
    for(int j = 0; j < indexes.length; j++) {
      lineConvert.convertFloat(output, indexes[j], values[j]);
    }
  }
}

private String toString(Int2IntOpenHashMap clocVec) {
  if(clocVec == null) {
    return "NULL";
  }

  StringBuilder sb = new StringBuilder();
  ObjectIterator<Int2IntMap.Entry> iter = clocVec.int2IntEntrySet().fastIterator();
  Int2IntMap.Entry item ;
  while(iter.hasNext()) {
    item = iter.next();
    sb.append(item.getIntKey());
    sb.append(":");
    sb.append(item.getIntValue());
    sb.append(";");
  }
  return sb.toString();
}

/**
 * Get task clocks for all matrices from Master
 * @return task clocks for all matrices from Master
 * @throws ServiceException
 */
public Int2ObjectOpenHashMap<Int2IntOpenHashMap> getTaskMatrixClocks() throws ServiceException {
  GetTaskMatrixClockResponse response = masterProxy.getTaskMatrixClocks(null,
    GetTaskMatrixClockRequest.newBuilder().build());
  Int2ObjectOpenHashMap<Int2IntOpenHashMap> taskIdToMatrixClocksMap = new Int2ObjectOpenHashMap<>(response.getTaskMatrixClocksCount());

  List<TaskMatrixClock> taskMatrixClocks = response.getTaskMatrixClocksList();
  int size = taskMatrixClocks.size();
  int matrixNum;
  for(int i = 0; i < size; i++) {
    Int2IntOpenHashMap matrixIdToClockMap = new Int2IntOpenHashMap(taskMatrixClocks.get(i).getMatrixClocksCount());
    taskIdToMatrixClocksMap.put(taskMatrixClocks.get(i).getTaskId().getTaskIndex(), matrixIdToClockMap);
    List<MatrixClock> matrixClocks = taskMatrixClocks.get(i).getMatrixClocksList();
    matrixNum = matrixClocks.size();
    for(int j = 0; j < matrixNum; j++) {
      matrixIdToClockMap.put(matrixClocks.get(j).getMatrixId(), matrixClocks.get(j).getClock());
    }
  }

  return taskIdToMatrixClocksMap;
}

/**
 * Set clock vector
 * @param clockVec clock vector
 */
public void setClockVec(Int2IntOpenHashMap clockVec) {
  try {
    lock.writeLock().lock();
    ObjectIterator<Int2IntMap.Entry> iter = clockVec.int2IntEntrySet().fastIterator();
    Int2IntMap.Entry item;
    while(iter.hasNext()) {
      item = iter.next();
      if(!taskIndexToClockMap.containsKey(item.getIntKey())
        || (taskIndexToClockMap.containsKey(item.getIntKey())
        && taskIndexToClockMap.get(item.getIntKey()) < item.getIntValue())) {
        taskIndexToClockMap.put(item.getIntKey(), item.getIntValue());
      }
    }
    refreshMinClock();
  } finally {
    lock.writeLock().unlock();
  }
}

/**
 * Adjust clock values
 * @param taskToMatrixClocks taskId->(matrixId->clock) map
 */
public void adjustClocks(Int2ObjectOpenHashMap<Int2IntOpenHashMap> taskToMatrixClocks) {
  ObjectIterator<Int2ObjectMap.Entry<Int2IntOpenHashMap>> taskIter =
    taskToMatrixClocks.int2ObjectEntrySet().fastIterator();
  Int2ObjectMap.Entry<Int2IntOpenHashMap> taskEntry = null;
  int taskId = 0;
  Int2IntOpenHashMap matrixIdToClockMap = null;
  ObjectIterator<Int2IntMap.Entry> matrixIter = null;
  Int2IntMap.Entry matrixEntry = null;

  while(taskIter.hasNext()) {
    taskEntry = taskIter.next();
    taskId = taskEntry.getIntKey();
    matrixIdToClockMap = taskEntry.getValue();
    matrixIter = matrixIdToClockMap.int2IntEntrySet().fastIterator();
    while (matrixIter.hasNext()) {
      matrixEntry = matrixIter.next();
      updateClock(matrixEntry.getIntKey(), taskId, matrixEntry.getIntValue());
    }
  }
}

/**
 * get clock of all matrices for all task
 * @param controller rpc controller of protobuf
 * @param request contains task id
 * @throws ServiceException
 */
@Override
public GetTaskMatrixClockResponse getTaskMatrixClocks(RpcController controller,
    GetTaskMatrixClockRequest request) throws ServiceException {
  AMTaskManager taskManager = context.getTaskManager();
  Collection<AMTask> tasks = taskManager.getTasks();
  GetTaskMatrixClockResponse.Builder builder = GetTaskMatrixClockResponse.newBuilder();
  TaskMatrixClock.Builder taskBuilder = TaskMatrixClock.newBuilder();
  MatrixClock.Builder matrixClockBuilder = MatrixClock.newBuilder();

  Int2IntOpenHashMap matrixClocks = null;
  for(AMTask task:tasks){
    taskBuilder.setTaskId(ProtobufUtil.convertToIdProto(task.getTaskId()));
    matrixClocks = task.getMatrixClocks();
    for(it.unimi.dsi.fastutil.ints.Int2IntMap.Entry entry:matrixClocks.int2IntEntrySet()) {
      taskBuilder.addMatrixClocks(matrixClockBuilder.setMatrixId(entry.getIntKey()).setClock(entry.getIntValue()).build());
    }
    builder.addTaskMatrixClocks(taskBuilder.build());
    taskBuilder.clear();
  }

  return builder.build();
}

public AMTask(TaskId id, AMTask amTask) {
  state = AMTaskState.NEW;
  taskId = id;
  metrics = new HashMap<String, String>();
  startTime = -1;
  finishTime = -1;

  matrixIdToClockMap = new Int2IntOpenHashMap();
  // if amTask is not null, we should clone task state from it
  if (amTask == null) {
    iteration = 0;
    progress = 0.0f;
  } else {
    iteration = amTask.getIteration();
    progress = amTask.getProgress();
    matrixIdToClockMap.putAll(amTask.matrixIdToClockMap);
  }

  ReadWriteLock readWriteLock = new ReentrantReadWriteLock();
  readLock = readWriteLock.readLock();
  writeLock = readWriteLock.writeLock();
}

@Test
public void testUpdateIntSparseToIntSparse() throws Exception {
  ServerSparseIntRow serverSparseIntRow = new ServerSparseIntRow(rowId, startCol, endCol);
  ByteBuf buf = Unpooled.buffer(16);
  buf.writeInt(0);
  buf.writeInt(0);
  buf.writeInt(1);
  buf.writeInt(1);
  buf.writeInt(2);
  buf.writeInt(2);
  rowUpdater.updateIntSparseToIntSparse(3, buf, serverSparseIntRow);
  Int2IntOpenHashMap hashMap = new Int2IntOpenHashMap();
  hashMap.addTo(0, 0);
  hashMap.addTo(1, 1);
  hashMap.addTo(2, 2);
  assertEquals(serverSparseIntRow.getData(), hashMap);
}

private void fillSeenPositions(Collection<DexMethod> invokes) {
  for (DexMethod method : invokes) {
    DexType[] parameters = method.proto.parameters.values;
    int arity = parameters.length;
    int positions = computePositionsFor(method.proto, target, targetProtoCache, substituions);
    if (positions != 0) {
      Int2IntMap positionsMap =
          seenPositions.computeIfAbsent(method.name, k -> {
            Int2IntMap result = new Int2IntOpenHashMap();
            result.defaultReturnValue(NOT_FOUND);
            return result;
          });
      int value = 0;
      int previous = positionsMap.get(arity);
      if (previous != NOT_FOUND) {
        value = previous;
      }
      value |= positions;
      positionsMap.put(arity, value);
    }
  }

}

/**
 */
public Table countByCategory() {
    Table t = new Table("Column: " + name());
    CategoryColumn categories = new CategoryColumn("Category");
    IntColumn counts = new IntColumn("Count");

    Int2IntMap valueToCount = new Int2IntOpenHashMap();
    for (int next : values) {
        if (valueToCount.containsKey(next)) {
            valueToCount.put(next, valueToCount.get(next) + 1);
        } else {
            valueToCount.put(next, 1);
        }
    }

    for (Map.Entry<Integer, Integer> entry : valueToCount.int2IntEntrySet()) {
        categories.add(lookupTable.get(entry.getKey()));
        counts.append(entry.getValue());
    }
    t.addColumn(categories);
    t.addColumn(counts);
    return t;
}

/**
 * This starts the graph off with a single segment, and additional ones are allocated as needed.
 *
 * @param maxNumSegments                 is the maximum number of segments we'll add to the graph.
 *                                       At that point, the oldest segments will start getting
 *                                       dropped
 * @param maxNumEdgesPerSegment          determines when the implementation decides to fork off a
 *                                       new segment
 * @param bipartiteGraphSegmentProvider  is used to generate new segments that are added to the
 *                                       graph
 * @param statsReceiver                  tracks the internal stats
 */
public LeftIndexedMultiSegmentBipartiteGraph(
    int maxNumSegments,
    int maxNumEdgesPerSegment,
    BipartiteGraphSegmentProvider<T> bipartiteGraphSegmentProvider,
    MultiSegmentReaderAccessibleInfoProvider<T> multiSegmentReaderAccessibleInfoProvider,
    StatsReceiver statsReceiver) {
  this.maxNumSegments = maxNumSegments;
  this.maxNumEdgesPerSegment = maxNumEdgesPerSegment;
  this.bipartiteGraphSegmentProvider = bipartiteGraphSegmentProvider;
  this.statsReceiver = statsReceiver.scope("LeftIndexedMultiSegmentBipartiteGraph");
  this.numEdgesSeenInAllHistoryCounter = this.statsReceiver.counter("numEdgesSeenInAllHistory");
  this.multiSegmentReaderAccessibleInfoProvider = multiSegmentReaderAccessibleInfoProvider;
  this.numEdgesInNonLiveSegmentsMap = new Int2IntOpenHashMap(maxNumSegments);
  addNewSegment();
}

private Int2IntMap getFasterIntersectionMap(int uidx) {
    Int2IntOpenHashMap intersectionMap = new Int2IntOpenHashMap();
    intersectionMap.defaultReturnValue(0);

    IntIterator iidxs = data.getUidxIidxs(uidx);
    while (iidxs.hasNext()) {
        IntIterator vidxs = data.getIidxUidxs(iidxs.nextInt());
        while (vidxs.hasNext()) {
            intersectionMap.addTo(vidxs.nextInt(), 1);
        }
    }

    intersectionMap.remove(uidx);

    return intersectionMap;
}

/**
 */
public Table countByCategory() {
    Table t = new Table("Column: " + name());
    CategoryColumn categories = new CategoryColumn("Category");
    IntColumn counts = new IntColumn("Count");

    Int2IntMap valueToCount = new Int2IntOpenHashMap();
    for (int next : values) {
        if (valueToCount.containsKey(next)) {
            valueToCount.put(next, valueToCount.get(next) + 1);
        } else {
            valueToCount.put(next, 1);
        }
    }

    for (Map.Entry<Integer, Integer> entry : valueToCount.int2IntEntrySet()) {
        categories.add(lookupTable.get(entry.getKey()));
        counts.append(entry.getValue());
    }
    t.addColumn(categories);
    t.addColumn(counts);
    return t;
}

private void addCounts(Int2IntOpenHashMap phraseToClusterMap, Int2IntOpenHashMap phraseContextCounts, Map<Integer, Int2IntOpenHashMap> prevClusterCounts, Map<Integer, Int2IntOpenHashMap> nextClusterCounts, int phrase, boolean includeIdentityCounts, int newCluster) {
    Int2IntOpenHashMap phrasePrevClusterCounts = prevClusterCounts.get(newCluster);
    if (phrasePrevClusterCounts == null) {
        phrasePrevClusterCounts = ContextCountsUtils.createNewInt2IntMap();
        prevClusterCounts.put(newCluster, phrasePrevClusterCounts);
    }
    for (Int2IntOpenHashMap.Entry otherPhraseEntry : phraseContextCounts.int2IntEntrySet()) {
        int otherPhrase = otherPhraseEntry.getIntKey();
        if (phrase != otherPhrase || includeIdentityCounts) {
            int clusterOtherPhrase = otherPhrase == phrase ? newCluster : phraseToClusterMap.get(otherPhrase);
            phrasePrevClusterCounts.addTo(clusterOtherPhrase, otherPhraseEntry.getIntValue());
            Int2IntOpenHashMap otherPhraseNextCounts = nextClusterCounts.get(clusterOtherPhrase);
            if (otherPhraseNextCounts == null) {
                otherPhraseNextCounts = ContextCountsUtils.createNewInt2IntMap();
                nextClusterCounts.put(clusterOtherPhrase, otherPhraseNextCounts);
            }
            otherPhraseNextCounts.addTo(newCluster, otherPhraseEntry.getValue());
        }
    }
}

protected void mergeCounts(int smallCluster, int largeCluster, Map<Integer, Int2IntOpenHashMap> counts) {
    //step 1: merge counts from small cluster to large cluster
    Int2IntOpenHashMap countsSmallCluster = counts.remove(smallCluster);
    Int2IntOpenHashMap countsLargeCluster = counts.get(largeCluster);
    if (countsLargeCluster == null) {
        countsLargeCluster = ContextCountsUtils.createNewInt2IntMap();
        counts.put(largeCluster, countsLargeCluster);
    }
    for (Int2IntOpenHashMap.Entry entry : countsSmallCluster.int2IntEntrySet()) {
        countsLargeCluster.addTo(entry.getIntKey(), entry.getIntValue());
    }
    //step 2: update all occurrences of small cluster to the large cluster
    counts.values().parallelStream().forEach(countsForSingleCluster -> {
        int prevCountsSmallCluster = countsForSingleCluster.remove(smallCluster);
        if (prevCountsSmallCluster > 0) {
            countsForSingleCluster.addTo(largeCluster, prevCountsSmallCluster);
        }
    });
}

private void mapCluster(Map<Integer, Int2IntOpenHashMap> counts, int oldCluster, int newCluster) {
    Int2IntOpenHashMap mapOldCluster = counts.remove(oldCluster);
    if (mapOldCluster != null) {
        Int2IntOpenHashMap mapNewCluster = counts.get(newCluster);
        if (mapNewCluster == null) {
            counts.put(newCluster, mapOldCluster);
        } else {
            //merge maps
            for (Map.Entry<Integer, Integer> entry : mapOldCluster.entrySet()) {
                mapNewCluster.add(entry.getKey(), entry.getValue());
            }
        }
    }
    for (Int2IntOpenHashMap currMap : counts.values()) {
        currMap.addTo(newCluster, currMap.remove(oldCluster));
    }
}

@Override
public Kryo create() {
  Kryo kryo = new Kryo();
  kryo.setRegistrationRequired(true);

  // model class(es)
  kryo.register(LookupUsage.class);

  // fastutils
  kryo.register(Int2IntArrayMap.class);
  kryo.register(Int2IntOpenHashMap.class);

  // java & commons
  kryo.register(Date.class);
  kryo.register(HashMap.class);
  kryo.register(HashSet.class);
  kryo.register(ArrayList.class);
  ImmutableListSerializer.registerSerializers(kryo);

  // enums
  kryo.register(Rank.class);
  kryo.register(Kingdom.class);

  return kryo;
}

/**
 * Get a model row
 *
 * @param rowId row index
 * @param partId partition index
 * @return model row
 */
public Int2IntOpenHashMap getRow(int rowId, int partId) {
  synchronized (this) {
    if (tempModel.get(rowId) == null) {
      tempModel.put(rowId, new HashMap<>());
      tempModel.get(rowId).put(partId, new Int2IntOpenHashMap());
    } else {
      if (tempModel.get(rowId).get(partId) == null) {
        tempModel.get(rowId).put(partId, new Int2IntOpenHashMap());
      }
    }

    return tempModel.get(rowId).get(partId);
  }
}

/**
 * Get a model row
 *
 * @param rowId row index
 * @param partId partition index
 * @param nnz estimated non-zero number zero number
 * @return model row
 */
public Int2IntOpenHashMap getRow(int rowId, int partId, int nnz) {
  synchronized (this) {
    if (tempModel.get(rowId) == null) {
      tempModel.put(rowId, new HashMap<>());
      tempModel.get(rowId).put(partId, new Int2IntOpenHashMap(nnz));
    } else {
      if (tempModel.get(rowId).get(partId) == null) {
        tempModel.get(rowId).put(partId, new Int2IntOpenHashMap(nnz));
      }
    }

    return tempModel.get(rowId).get(partId);
  }
}

public static Int2IntOpenHashMap loadSparseIntRowFromPartition(FSDataInputStream input,
    ModelPartitionMeta partMeta, int rowId) throws IOException {
  RowOffset rowOffset = partMeta.getRowMetas().get(rowId);
  input.seek(rowOffset.getOffset());
  Preconditions.checkState (input.readInt() == rowId);
  int num = input.readInt();
  Int2IntOpenHashMap row = new Int2IntOpenHashMap();
  for (int i = 0; i < num; i++) {
    row.put(input.readInt(), input.readInt());
  }
  return row;
}

@Override
public void deserialize(ByteBuf buf) {
  int dim = buf.readInt();
  int length = buf.readInt();
  Int2IntOpenHashMap data = new Int2IntOpenHashMap(length);
  IntStream.range(0,length).forEach(i-> data.put(buf.readInt(), buf.readInt()));
  this.dim = dim;
  this.hashMap = data;
}

@Override
public void deserialize(ByteBuf buf) {
  super.deserialize(buf);
  part = new ServerPartition();
  part.deserialize(buf);
  int clockVecSize = buf.readInt();
  if(clockVecSize > 0) {
    taskIndexToClockMap = new Int2IntOpenHashMap(clockVecSize);
    for(int i = 0; i < clockVecSize; i++) {
      taskIndexToClockMap.put(buf.readInt(), buf.readInt());
    }
  }
}

private boolean isClockReady(PartitionKey partKey, int clock) {
  boolean ready = clock < 0 || context.getClockVectorManager().getPartClock(partKey.getMatrixId(), partKey.getPartitionId()) >= clock;
  if(!ready) {
    try {
      Int2ObjectOpenHashMap<Int2IntOpenHashMap> clocks = context.getMaster().getTaskMatrixClocks();
      context.getClockVectorManager().adjustClocks(clocks);
    } catch (ServiceException e) {
      LOG.error("Get Clocks from master falied,", e);
    }
    ready = clock < 0 || context.getClockVectorManager().getPartClock(partKey.getMatrixId(), partKey.getPartitionId()) >= clock;
  }
  return ready;
}

/**
 * Recover a partition
 * @param seqId rpc request it
 * @param request request
 * @return serialized rpc response
 */
private ByteBuf recoverPart(int seqId, RecoverPartRequest request) {
  if (LOG.isDebugEnabled()) {
    LOG.debug("recover part request=" + request + " with seqId=" + seqId);
  }

  long startTs = System.currentTimeMillis();
  ByteBuf buf = ByteBufUtils.newByteBuf(8 + 4);
  buf.writeInt(seqId);

  Response response = null;

  PartitionKey partKey = request.getPartKey();
  Int2IntOpenHashMap clockVec = request.getTaskIndexToClockMap();
  if(clockVec != null) {
    context.getClockVectorManager().setClockVec(partKey.getMatrixId(), partKey.getPartitionId(), clockVec);
  }

  ServerPartition part = context.getMatrixStorageManager().getPart(partKey.getMatrixId(), partKey.getPartitionId());
  if(part == null) {
    String log = "can not find the partition " + partKey;
    response = new Response(ResponseType.SERVER_HANDLE_FATAL, log);
    response.serialize(buf);
    return buf;
  }
  part.recover(request.getPart());
  response = new Response(ResponseType.SUCCESS);
  response.serialize(buf);
  if (LOG.isDebugEnabled()) {
    LOG.debug(
      "recover partition  request " + request + " use time=" + (System.currentTimeMillis()
        - startTs));
  }

  return buf;
}

/**
 * Get partition id to partition clock map
 * @return partition id to partition clock map
 */
public Int2IntOpenHashMap getPartClocks() {
  Int2IntOpenHashMap partClocks = new Int2IntOpenHashMap(partIdToClockVecMap.size());
  for(Map.Entry<Integer, PartClockVector> entry : partIdToClockVecMap.entrySet()) {
    partClocks.put(entry.getKey().intValue(), entry.getValue().getMinClock());
  }
  return partClocks;
}

/**
 * Create a PartClockVector
 * @param taskNum total task number
 */
public PartClockVector(int taskNum) {
  this.taskNum = taskNum;
  minClock = 0;
  taskIndexToClockMap = new Int2IntOpenHashMap(taskNum);
  for(int i = 0; i < taskNum; i++) {
    taskIndexToClockMap.put(i, 0);
  }
  lock = new ReentrantReadWriteLock();
}

/**
 * Get clock vector
 * @return clock vector
 */
public Int2IntOpenHashMap getClockVec() {
  try {
    lock.readLock().lock();
    return taskIndexToClockMap.clone();
  } finally {
    lock.readLock().unlock();
  }
}

/**
 * Get clock vector of a matrix partition
 * @param matrixId matrix id
 * @param partId partition id
 * @return clock vector
 */
public Int2IntOpenHashMap getClockVec(int matrixId, int partId) {
  MatrixClockVector matrixClockVector = matrixIdToClockVecMap.get(matrixId);
  if(matrixClockVector == null) {
    LOG.warn("get clock vector for a non-exist matrix " + matrixId);
    return new Int2IntOpenHashMap();
  }
  return matrixClockVector.getClockVec(partId);
}

/**
 * Get partition clocks for a matrix
 * @param matrixId matrix id
 * @return partition clocks
 */
public Int2IntOpenHashMap getPartClocks(int matrixId) {
  MatrixClockVector matrixClockVector = matrixIdToClockVecMap.get(matrixId);
  if(matrixClockVector == null) {
    LOG.warn("get clock vector for a non-exist matrix " + matrixId);
    return new Int2IntOpenHashMap();
  }
  return matrixClockVector.getPartClocks();
}

/**
 * Set partition clock vector
 * @param matrixId matrix id
 * @param partId partition id
 * @param clockVec clock vector
 */
public void setClockVec(int matrixId, int partId, Int2IntOpenHashMap clockVec) {
  MatrixClockVector matrixClockVector = matrixIdToClockVecMap.get(matrixId);
  if(matrixClockVector == null) {
    LOG.warn("update clock vector for a non-exist matrix " + matrixId);
    return;
  }
  matrixClockVector.setClockVec(partId, clockVec);
}

/**
 * Merge this sparse int vector split to a map
 * @param indexToValueMap a index->value map
 */
public void mergeTo(Int2IntOpenHashMap indexToValueMap) {
  try {
    lock.readLock().lock();
    indexToValueMap.putAll(hashMap);
  } finally {
    lock.readLock().unlock();
  }
}

private static WorkerMetaInfoProto buildWorkerMetaProto(AMWorker worker) {
  WorkerMetaInfoProto.Builder builder = WorkerMetaInfoProto.newBuilder();

  WorkerAttempt attempt = worker.getRunningAttempt();
  WorkerAttemptIdProto workerAttemptIdProto = convertToIdProto(attempt.getId());
  Location location = attempt.getLocation();

  WorkerLocationProto.Builder locBuilder = WorkerLocationProto.newBuilder();
  locBuilder.setWorkerAttemptId(workerAttemptIdProto);
  if (location != null) {
    locBuilder.setLocation(buildLocation(location));
  }
  builder.setWorkerLocation(locBuilder.build());

  TaskMetaInfoProto.Builder taskMetaBuilder = TaskMetaInfoProto.newBuilder();
  MatrixClock.Builder clockBuilder = MatrixClock.newBuilder();
  for (Entry<TaskId, AMTask> taskEntry : attempt.getTaskMap().entrySet()) {
    AMTask task = taskEntry.getValue();
    taskMetaBuilder.setTaskId(convertToIdProto(taskEntry.getKey()));
    taskMetaBuilder.setIteration(task.getIteration());

    Int2IntOpenHashMap matrixClocks = task.getMatrixClocks();
    for (it.unimi.dsi.fastutil.ints.Int2IntMap.Entry clockEntry : matrixClocks
        .int2IntEntrySet()) {
      taskMetaBuilder.addMatrixClock(clockBuilder.setMatrixId(clockEntry.getIntKey())
          .setClock(clockEntry.getIntValue()).build());
    }
    builder.addTasks(taskMetaBuilder.build());
    LOG.debug("task meta=" + taskMetaBuilder.build());
  }

  return builder.build();
}

/**
 * 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);
}

/**
 * 
 * Create a new GetRowsFlowCache.
 *
 * @param totalRequestNum total sub-requests number
 * @param matrixId matrix id
 * @param rowIndexToPartSizeCache row index to the number of partitions that contain this row map
 */
public GetRowsFlowCache(int totalRequestNum, int matrixId,
    Int2IntOpenHashMap rowIndexToPartSizeCache) {
  super(totalRequestNum);
  this.matrixId = matrixId;
  this.rowIndexToPartSizeCache = rowIndexToPartSizeCache;
  rowsSplitCache = new Int2ObjectOpenHashMap<List<ServerRow>>();
  rowsSplitFutures = new ObjectOpenHashSet<Future<List<ServerRow>>>(totalRequestNum);
}

public MatrixOpLogCache() {
  opLogs = new ConcurrentHashMap<>();

  messageQueue = new PriorityBlockingQueue<OpLogMessage>(100, new PriorityComparator());
  seqIdToMessageMaps = new Int2ObjectOpenHashMap<Int2ObjectAVLTreeMap<OpLogMessage>>();
  waitedMessageQueues = new Int2ObjectOpenHashMap<LinkedBlockingQueue<OpLogMessage>>();
  flushListeners = new Int2ObjectOpenHashMap<List<OpLogMessage>>();
  seqIdGenerator = new AtomicInteger(0);
  mergingCounters = new Int2IntOpenHashMap();
  stopped = new AtomicBoolean(false);
  messageToFutureMap = new ConcurrentHashMap<OpLogMessage, Future<VoidResult>>();
}

/**
 * get all matrix clocks
 * @return Int2IntOpenHashMap  all matrix clocks
 */
public Int2IntOpenHashMap getMatrixClocks() {
  try {
    readLock.lock();
    return matrixIdToClockMap.clone();
  } finally {
    readLock.unlock();
  }
}

@Test
public void testUpdateIntDenseToIntSparse() throws Exception {
  ServerSparseIntRow serverSparseIntRow = new ServerSparseIntRow(rowId, startCol, endCol);
  ByteBuf buf = Unpooled.buffer(16);
  buf.writeInt(0);
  buf.writeInt(1);
  buf.writeInt(2);
  rowUpdater.updateIntDenseToIntSparse(3, buf, serverSparseIntRow);
  Int2IntOpenHashMap hashMap = new Int2IntOpenHashMap();
  hashMap.addTo(0, 0);
  hashMap.addTo(1, 1);
  hashMap.addTo(2, 2);
  assertEquals(serverSparseIntRow.getData(), hashMap);
}