HDDS-11209. Avoid insufficient EC pipelines in the container pipeline cache

[whbing]

What changes were proposed in this pull request?

OM cached pipeline may not cache container's all datanode information (for example, some DNs may not report in time), which can be problematic for EC files. If the nodes num is less than the EC data-blocks num, an error will occur when reading data:

There are insufficient datanodes to read the EC block 

This err will persist until the cache expires (after 6 hours) even if all DNs corresponding to the EC file are completely normal.

So, it should not cache EC pipelines with incomplete data-nodes.

What is the link to the Apache JIRA

https://issues.apache.org/jira/browse/HDDS-11209

How was this patch tested?

• unit tests,
• and manual tests

[errose28]

It looks like the code will also cache incomplete Ratis pipelines, just not those that are empty. While the reads will technically still work, this could cause under-replicated containers to be sitting in the cache for a while, and may skew the read load for some time even after replication completes. Should we make the cache more strict to only keep pipelines that have all nodes present?

There are insufficient datanodes to read the EC block
This err will persist until the cache expires (after 6 hours) even if all DNs corresponding to the EC file are completely normal.

The client should force a cache refresh on read failures like this or corrupted data since SCM likely made new copies since the last cache refresh. It looks like that is not happening which is another problem.

cc @duongkame

[whbing]

Should we make the cache more strict to only keep pipelines that have all nodes present?

@errose28 Sorry for late reply. I am inclined to make the cache more strict. At the very least, EC pipelines that do not have enough datanodes for reconstruction should not be cached, as they are invalid.
I believe the following approaches can be considered:

1. datanodes < data num : uncache
   datanodes >= data num: cache
2. datanodes < data num + parity num: uncache
   datanodes = data num + parity num: cache
3. datanodes loss data index: uncache
   datanodes contain all data index: cache

The current submission is based on point 3.

---

The client should force a cache refresh on read failures like this or corrupted data since SCM likely made new copies since the last cache refresh. It looks like that is not happening which is another problem.

Now no retry when InsufficientLocationsException. I tested some cases, such as incorrect locations (cached expired locations), all of these can retry with forceFreshPipeline = true.
Perhaps in the current, we can just deal with the case InsufficientLocationsException. Of course, it can be processed in another ticket.

[siddhantsangwan]

@whbing thanks for working on this.

datanodes loss data index: uncache
datanodes contain all data index: cache

I like this approach. Will review the code soon.

OM cached pipeline may not cache container's all datanode information (for example, some DNs may not report in time), which can be problematic for EC files.

This makes me wonder if the scm safe-mode rule for EC containers should be stricter than just seeing 1 replica.

[errose28]

This makes me wonder if the scm safe-mode rule for EC containers should be stricter than just seeing 1 replica.

@siddhantsangwan see #7008

[siddhantsangwan]

Looks good so far. @whbing can you please add a test?

[whbing]

please add a test?

@siddhantsangwan Added test for refreshing container when missing ec data indexes.

[siddhantsangwan]

@whbing It's worth checking for thread-safety issues here. I see that the cache is documented as thread safe, but the new map that we're using is not thread safe.

[siddhantsangwan]

Here, we're writing to the map in a multi-threaded context. We should either use a concurrent hash map or change this logic such that it's read-only.

[whbing]

@siddhantsangwan Thanks for suggestion ! It should fixed to concurrent hash map.

Btw, I performed benchmarking for several methods (as following). I think using map is unnecessary here and alternative approaches instead.

[siddhantsangwan]

Suggested change

	// case1: pipeline replicaIndexes missing some data indexes, not cache
	// case1: pipeline replicaIndexes missing some data indexes, should not cache

[siddhantsangwan]

Suggested change

	// case2: pipeline replicaIndexes contain all data indexes, cache
	// case2: pipeline replicaIndexes contain all data indexes, should cache

[whbing]

  private static final Map<Integer, Set<Integer>> EC_DATA_INDEXES = new ConcurrentHashMap<>();
  private static final int D = 6;
  private List<Integer> replicaIndexes;

  @Setup
  public void setup() {
    EC_DATA_INDEXES.clear();
    replicaIndexes = IntStream.rangeClosed(1, 9).boxed().collect(Collectors.toList());
    replicaIndexes.add(1);
    Collections.shuffle(replicaIndexes);
  }
 @Benchmark
  @Threads(500)
  public boolean testComputeIfAbsentAndContainsAll() {
    Set<Integer> dataIndexes = EC_DATA_INDEXES.computeIfAbsent(D, k ->
        IntStream.rangeClosed(1, k).boxed().collect(Collectors.toSet()));
    return !replicaIndexes.containsAll(dataIndexes);
  }

  @Benchmark
  @Threads(500)
  public boolean testDirectPutAndContainsAll() {
    Set<Integer> dataIndexes = IntStream.rangeClosed(1, D).boxed().collect(Collectors.toSet());
    return !replicaIndexes.containsAll(dataIndexes);
  }

  @Benchmark
  @Threads(500)
  public boolean testIndexCompare() {
    List<Integer> indexes = replicaIndexes.stream()
        .sorted()
        .collect(Collectors.toList());
    return !(indexes.size() >= D && indexes.get(D - 1) == D);
  }

Benchmark                                         Mode  Cnt  Score   Error  Units
ECDataIndexesBenchmark.testDirectContainsAll      avgt    4  0.029 ± 0.015  ms/op
ECDataIndexesBenchmark.testFromMapAndContainsAll  avgt    4  0.062 ± 0.023  ms/op
ECDataIndexesBenchmark.testIndexCompare           avgt    4  0.039 ± 0.017  ms/op

I tested that using map didn't seem to yield much benefit, so used another more concise method instead.

[whbing]

@siddhantsangwan Could you have another look? cc @duongkame @errose28

[siddhantsangwan]

@whbing thanks for the update and benchmark. I agree, the map introduces a bit more complexity than necessary. Have you also considered another simple approach - using a nested loop to check if the index is present?

      List<Long> uncachePipelines = result.entrySet().stream()
          .filter(e -> {
            Pipeline pipeline = e.getValue();
            // filter empty pipelines
            if (pipeline.isEmpty()) {
              return true;
            }
            // filter insufficient EC pipelines which missing any data index
            ReplicationConfig repConfig = pipeline.getReplicationConfig();
            if (repConfig instanceof ECReplicationConfig) {
              int d = ((ECReplicationConfig) repConfig).getData();
//              List<Integer> indexes = pipeline.getReplicaIndexes().values().stream()
//                  .sorted()
//                  .distinct()
//                  .collect(Collectors.toList());
//              return !(indexes.size() >= d && indexes.get(d - 1) == d);
              for (int i = 1; i <= d; i++) { // another approach
                if (!pipeline.getReplicaIndexes().containsValue(i)) {
                  return true;
                }
              }
            }
            return false;
          })
          .map(Map.Entry::getKey)
          .collect(Collectors.toList());

The tests you wrote pass with this approach. We expect the number of indexes for EC to be less than 20, so this should barely take any time and is also thread-safe. Let me know if you see any problems with this.

[siddhantsangwan]

@whbing thanks for the update. Is it ready for review?

[whbing]

Is it ready for review?

@siddhantsangwan Thanks! it's ready.

Let me know if you see any problems with this.

No problem! The giving method is more concise and efficient.

  @Benchmark
  @Threads(500)
  public boolean testIndexCompare() {
    List<Integer> indexes = replicaIndexes.stream()
        .sorted()
        .distinct()
        .collect(Collectors.toList());
    return !(indexes.size() >= D && indexes.get(D - 1) == D);
  }

  @Benchmark
  @Threads(500)
  public boolean testLoop() {
    for (int i = 1; i <= D; i++) {
      if (replicaIndexes.contains(i)) {
        return true;
      }
    }
    return false;
  }
Benchmark                                 Mode  Cnt  Score   Error  Units
ECDataIndexesBenchmark1.testIndexCompare  avgt    4  0.035 ± 0.035  ms/op
ECDataIndexesBenchmark1.testLoop          avgt    4  0.001 ± 0.001  ms/op

[siddhantsangwan]

No problem! The giving method is more concise and efficient.

Ah, that's good. Thanks for benchmarking this as well. LGTM, pending green CI.

[whbing]

Thanks @errose28 and @siddhantsangwan for view, thanks @siddhantsangwan for merge!