mirror of https://github.com/hyperledger/besu
Pipeline Improvements (#1117)
* Capture input and output metrics for each pipe rather than out input. * Replace the batching processor with a BatchingReadPipe that wraps a normal pipe and creates batching on the fly. Avoids needing an extra thread and synchronization overhead to create batches. Signed-off-by: Adrian Sutton <adrian.sutton@consensys.net>pull/2/head
Adrian Sutton
6 years ago
committed by
GitHub
parent
8ef56242dd
commit
30ec5b7ba1
@ -1,32 +0,0 @@ |
||||
/* |
||||
* Copyright 2019 ConsenSys AG. |
||||
* |
||||
* Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with |
||||
* the License. You may obtain a copy of the License at |
||||
* |
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
* |
||||
* Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on |
||||
* an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the |
||||
* specific language governing permissions and limitations under the License. |
||||
*/ |
||||
package tech.pegasys.pantheon.services.pipeline; |
||||
|
||||
import java.util.List; |
||||
|
||||
class BatchingProcessor<T> implements Processor<T, List<T>> { |
||||
|
||||
private final int maximumBatchSize; |
||||
|
||||
public BatchingProcessor(final int maximumBatchSize) { |
||||
this.maximumBatchSize = maximumBatchSize; |
||||
} |
||||
|
||||
@Override |
||||
public void processNextInput(final ReadPipe<T> inputPipe, final WritePipe<List<T>> outputPipe) { |
||||
final List<T> batch = inputPipe.getBatch(maximumBatchSize); |
||||
if (!batch.isEmpty()) { |
||||
outputPipe.put(batch); |
||||
} |
||||
} |
||||
} |
||||
@ -0,0 +1,73 @@ |
||||
/* |
||||
* Copyright 2019 ConsenSys AG. |
||||
* |
||||
* Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with |
||||
* the License. You may obtain a copy of the License at |
||||
* |
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
* |
||||
* Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on |
||||
* an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the |
||||
* specific language governing permissions and limitations under the License. |
||||
*/ |
||||
package tech.pegasys.pantheon.services.pipeline; |
||||
|
||||
import tech.pegasys.pantheon.metrics.Counter; |
||||
|
||||
import java.util.ArrayList; |
||||
import java.util.Collection; |
||||
import java.util.List; |
||||
|
||||
public class BatchingReadPipe<T> implements ReadPipe<List<T>> { |
||||
|
||||
private final ReadPipe<T> input; |
||||
private final int maximumBatchSize; |
||||
private final Counter batchCounter; |
||||
|
||||
public BatchingReadPipe( |
||||
final ReadPipe<T> input, final int maximumBatchSize, final Counter batchCounter) { |
||||
this.input = input; |
||||
this.maximumBatchSize = maximumBatchSize; |
||||
this.batchCounter = batchCounter; |
||||
} |
||||
|
||||
@Override |
||||
public boolean hasMore() { |
||||
return input.hasMore(); |
||||
} |
||||
|
||||
@Override |
||||
public List<T> get() { |
||||
final T firstItem = input.get(); |
||||
if (firstItem == null) { |
||||
// Contract of get is to explicitly return null when no more items are available.
|
||||
// An empty list is not a suitable thing to return here.
|
||||
return null; |
||||
} |
||||
final List<T> batch = new ArrayList<>(); |
||||
batch.add(firstItem); |
||||
input.drainTo(batch, maximumBatchSize - 1); |
||||
batchCounter.inc(); |
||||
return batch; |
||||
} |
||||
|
||||
@Override |
||||
public List<T> poll() { |
||||
final List<T> batch = new ArrayList<>(); |
||||
input.drainTo(batch, maximumBatchSize); |
||||
if (batch.isEmpty()) { |
||||
// Poll has to return null if the pipe is empty
|
||||
return null; |
||||
} |
||||
batchCounter.inc(); |
||||
return batch; |
||||
} |
||||
|
||||
@Override |
||||
public void drainTo(final Collection<List<T>> output, final int maxElements) { |
||||
final List<T> nextBatch = poll(); |
||||
if (nextBatch != null) { |
||||
output.add(nextBatch); |
||||
} |
||||
} |
||||
} |
||||
@ -1,60 +0,0 @@ |
||||
/* |
||||
* Copyright 2019 ConsenSys AG. |
||||
* |
||||
* Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with |
||||
* the License. You may obtain a copy of the License at |
||||
* |
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
* |
||||
* Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on |
||||
* an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the |
||||
* specific language governing permissions and limitations under the License. |
||||
*/ |
||||
package tech.pegasys.pantheon.services.pipeline; |
||||
|
||||
import static java.util.Arrays.asList; |
||||
import static org.assertj.core.api.Assertions.assertThat; |
||||
import static org.mockito.Mockito.mock; |
||||
import static org.mockito.Mockito.verifyZeroInteractions; |
||||
import static tech.pegasys.pantheon.metrics.noop.NoOpMetricsSystem.NO_OP_COUNTER; |
||||
|
||||
import java.util.List; |
||||
|
||||
import org.junit.Test; |
||||
|
||||
public class BatchingProcessorTest { |
||||
|
||||
private final Pipe<Integer> inputPipe = new Pipe<>(10, NO_OP_COUNTER); |
||||
private final Pipe<List<Integer>> outputPipe = new Pipe<>(10, NO_OP_COUNTER); |
||||
private final BatchingProcessor<Integer> stage = new BatchingProcessor<>(3); |
||||
|
||||
@Test |
||||
public void shouldCreateBatches() { |
||||
for (int i = 1; i <= 8; i++) { |
||||
inputPipe.put(i); |
||||
} |
||||
inputPipe.close(); |
||||
|
||||
stage.processNextInput(inputPipe, outputPipe); |
||||
|
||||
assertThat(outputPipe.poll()).isEqualTo(asList(1, 2, 3)); |
||||
assertThat(outputPipe.poll()).isNull(); |
||||
|
||||
stage.processNextInput(inputPipe, outputPipe); |
||||
assertThat(outputPipe.poll()).isEqualTo(asList(4, 5, 6)); |
||||
assertThat(outputPipe.poll()).isNull(); |
||||
|
||||
stage.processNextInput(inputPipe, outputPipe); |
||||
assertThat(outputPipe.poll()).isEqualTo(asList(7, 8)); |
||||
assertThat(outputPipe.poll()).isNull(); |
||||
} |
||||
|
||||
@Test |
||||
public void shouldNotOutputItemWhenInputIsClosed() { |
||||
@SuppressWarnings("unchecked") |
||||
final WritePipe<List<Integer>> outputPipe = mock(WritePipe.class); |
||||
inputPipe.close(); |
||||
stage.processNextInput(inputPipe, outputPipe); |
||||
verifyZeroInteractions(outputPipe); |
||||
} |
||||
} |
||||
@ -0,0 +1,134 @@ |
||||
/* |
||||
* Copyright 2019 ConsenSys AG. |
||||
* |
||||
* Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with |
||||
* the License. You may obtain a copy of the License at |
||||
* |
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
* |
||||
* Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on |
||||
* an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the |
||||
* specific language governing permissions and limitations under the License. |
||||
*/ |
||||
package tech.pegasys.pantheon.services.pipeline; |
||||
|
||||
import static java.util.Arrays.asList; |
||||
import static org.assertj.core.api.Assertions.assertThat; |
||||
import static org.mockito.Mockito.mock; |
||||
import static org.mockito.Mockito.times; |
||||
import static org.mockito.Mockito.verify; |
||||
import static org.mockito.Mockito.verifyZeroInteractions; |
||||
import static tech.pegasys.pantheon.metrics.noop.NoOpMetricsSystem.NO_OP_COUNTER; |
||||
|
||||
import tech.pegasys.pantheon.metrics.Counter; |
||||
|
||||
import java.util.ArrayList; |
||||
import java.util.List; |
||||
|
||||
import org.junit.Test; |
||||
|
||||
public class BatchingReadPipeTest { |
||||
|
||||
private final Pipe<String> source = new Pipe<>(10, NO_OP_COUNTER, NO_OP_COUNTER); |
||||
private final Counter batchCounter = mock(Counter.class); |
||||
private final BatchingReadPipe<String> batchingPipe = |
||||
new BatchingReadPipe<>(source, 3, batchCounter); |
||||
|
||||
@Test |
||||
public void shouldGetABatchOfAvailableItems() { |
||||
source.put("a"); |
||||
source.put("b"); |
||||
source.put("c"); |
||||
source.put("d"); |
||||
|
||||
assertThat(batchingPipe.get()).containsExactly("a", "b", "c"); |
||||
} |
||||
|
||||
@Test |
||||
public void shouldNotWaitToFillBatch() { |
||||
source.put("a"); |
||||
assertThat(batchingPipe.get()).containsExactly("a"); |
||||
} |
||||
|
||||
@Test |
||||
public void shouldPollForNextBatchWhenAvailable() { |
||||
source.put("a"); |
||||
assertThat(batchingPipe.poll()).containsExactly("a"); |
||||
} |
||||
|
||||
@Test |
||||
public void shouldReturnNullFromPollWhenNoItemsAvailable() { |
||||
assertThat(batchingPipe.poll()).isNull(); |
||||
} |
||||
|
||||
@Test |
||||
public void shouldHaveMoreItemsWhileSourceHasMoreItems() { |
||||
assertThat(batchingPipe.hasMore()).isTrue(); |
||||
|
||||
source.put("a"); |
||||
source.put("b"); |
||||
source.put("c"); |
||||
|
||||
assertThat(batchingPipe.hasMore()).isTrue(); |
||||
|
||||
source.close(); |
||||
|
||||
assertThat(batchingPipe.hasMore()).isTrue(); |
||||
|
||||
assertThat(batchingPipe.get()).containsExactly("a", "b", "c"); |
||||
|
||||
assertThat(batchingPipe.hasMore()).isFalse(); |
||||
|
||||
assertThat(batchingPipe.get()).isNull(); |
||||
assertThat(batchingPipe.poll()).isNull(); |
||||
} |
||||
|
||||
@Test |
||||
public void shouldAddAtMostOneItemWhenDraining() { |
||||
// Collecting a batch of batches is pretty silly so only ever drain one batch at a time.
|
||||
source.put("a"); |
||||
source.put("b"); |
||||
source.put("c"); |
||||
source.put("1"); |
||||
source.put("2"); |
||||
source.put("3"); |
||||
|
||||
final List<List<String>> output = new ArrayList<>(); |
||||
batchingPipe.drainTo(output, 6); |
||||
// Note still only 3 items in the batch.
|
||||
assertThat(output).containsExactly(asList("a", "b", "c")); |
||||
} |
||||
|
||||
@Test |
||||
public void shouldCountBatchesReturnedFromGet() { |
||||
source.put("a"); |
||||
source.close(); |
||||
batchingPipe.get(); |
||||
assertThat(batchingPipe.get()).isNull(); |
||||
|
||||
verify(batchCounter, times(1)).inc(); |
||||
} |
||||
|
||||
@Test |
||||
public void shouldCountBatchesReturnedFromPoll() { |
||||
assertThat(batchingPipe.poll()).isNull(); |
||||
verifyZeroInteractions(batchCounter); |
||||
|
||||
source.put("a"); |
||||
batchingPipe.poll(); |
||||
|
||||
verify(batchCounter, times(1)).inc(); |
||||
} |
||||
|
||||
@Test |
||||
public void shouldCountBatchesReturnedFromDrainTo() { |
||||
final List<List<String>> output = new ArrayList<>(); |
||||
batchingPipe.drainTo(output, 3); |
||||
verifyZeroInteractions(batchCounter); |
||||
|
||||
source.put("a"); |
||||
batchingPipe.drainTo(output, 3); |
||||
|
||||
verify(batchCounter, times(1)).inc(); |
||||
} |
||||
} |
||||
Loading…
Reference in new issue