You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by pn...@apache.org on 2019/06/04 10:12:49 UTC
[flink] branch master updated (88c7d82 -> a36ec7f)
This is an automated email from the ASF dual-hosted git repository.
pnowojski pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git.
from 88c7d82 [FLINK-12604][table] Register a TableSource/Sink directly as CatalogTables
new d952e71 [hotfix][tests] Refactor the creation of InputGate for StreamNetworkBenchmarkEnvironment
new 3351ca1 [FLINK-12201][network,metrics] Introduce InputGateWithMetrics in Task to increment numBytesIn metric
new a36ec7f [hotfix][network] Drop redundant getSizeUnsafe from Buffer interface
The 3 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../runtime/io/network/NetworkEnvironment.java | 7 +-
.../flink/runtime/io/network/buffer/Buffer.java | 11 ---
.../runtime/io/network/buffer/NetworkBuffer.java | 5 -
.../buffer/ReadOnlySlicedNetworkBuffer.java | 5 -
.../network/partition/consumer/BufferOrEvent.java | 21 +++-
.../partition/consumer/LocalInputChannel.java | 2 +-
.../partition/consumer/RemoteInputChannel.java | 2 +-
.../partition/consumer/SingleInputGate.java | 9 +-
.../partition/consumer/SingleInputGateFactory.java | 5 +-
.../runtime/taskmanager/InputGateWithMetrics.java | 107 +++++++++++++++++++++
.../org/apache/flink/runtime/taskmanager/Task.java | 14 ++-
.../io/network/buffer/NetworkBufferTest.java | 6 --
.../network/buffer/ReadOnlySlicedBufferTest.java | 10 +-
.../network/partition/InputGateFairnessTest.java | 2 -
.../partition/consumer/SingleInputGateBuilder.java | 5 -
.../partition/consumer/SingleInputGateTest.java | 7 +-
.../flink/streaming/runtime/io/BufferSpiller.java | 4 +-
.../StreamNetworkBenchmarkEnvironment.java | 89 +++++++++--------
18 files changed, 193 insertions(+), 118 deletions(-)
create mode 100644 flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/InputGateWithMetrics.java
[flink] 01/03: [hotfix][tests] Refactor the creation of InputGate
for StreamNetworkBenchmarkEnvironment
Posted by pn...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
pnowojski pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit d952e710351e5a7c872f8b5379b1e8bab78ac9f6
Author: Zhijiang <wa...@aliyun.com>
AuthorDate: Wed May 29 23:07:48 2019 +0800
[hotfix][tests] Refactor the creation of InputGate for StreamNetworkBenchmarkEnvironment
---
.../StreamNetworkBenchmarkEnvironment.java | 79 +++++++++++-----------
1 file changed, 39 insertions(+), 40 deletions(-)
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/io/benchmark/StreamNetworkBenchmarkEnvironment.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/io/benchmark/StreamNetworkBenchmarkEnvironment.java
index 4b28961..f00a673 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/io/benchmark/StreamNetworkBenchmarkEnvironment.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/io/benchmark/StreamNetworkBenchmarkEnvironment.java
@@ -26,7 +26,6 @@ import org.apache.flink.runtime.clusterframework.types.ResourceID;
import org.apache.flink.runtime.deployment.InputChannelDeploymentDescriptor;
import org.apache.flink.runtime.deployment.InputGateDeploymentDescriptor;
import org.apache.flink.runtime.deployment.ResultPartitionLocation;
-import org.apache.flink.runtime.executiongraph.ExecutionAttemptID;
import org.apache.flink.runtime.io.disk.iomanager.IOManager;
import org.apache.flink.runtime.io.disk.iomanager.IOManagerAsync;
import org.apache.flink.runtime.io.network.ConnectionID;
@@ -77,7 +76,6 @@ public class StreamNetworkBenchmarkEnvironment<T extends IOReadableWritable> {
protected final JobID jobId = new JobID();
protected final IntermediateDataSetID dataSetID = new IntermediateDataSetID();
- protected final ExecutionAttemptID executionAttemptID = new ExecutionAttemptID();
protected NetworkEnvironment senderEnv;
protected NetworkEnvironment receiverEnv;
@@ -91,6 +89,8 @@ public class StreamNetworkBenchmarkEnvironment<T extends IOReadableWritable> {
private int dataPort;
+ private SingleInputGateFactory gateFactory;
+
public void setUp(
int writers,
int channels,
@@ -152,6 +152,13 @@ public class StreamNetworkBenchmarkEnvironment<T extends IOReadableWritable> {
receiverEnv.start();
}
+ gateFactory = new SingleInputGateFactory(
+ receiverEnv.getConfiguration(),
+ receiverEnv.getConnectionManager(),
+ receiverEnv.getResultPartitionManager(),
+ new TaskEventDispatcher(),
+ receiverEnv.getNetworkBufferPool());
+
generatePartitionIds();
}
@@ -167,12 +174,7 @@ public class StreamNetworkBenchmarkEnvironment<T extends IOReadableWritable> {
LOCAL_ADDRESS,
dataPort);
- InputGate receiverGate = createInputGate(
- dataSetID,
- executionAttemptID,
- senderLocation,
- receiverEnv,
- channels);
+ InputGate receiverGate = createInputGate(senderLocation);
SerializingLongReceiver receiver = new SerializingLongReceiver(receiverGate, channels * partitionIds.length);
@@ -228,40 +230,19 @@ public class StreamNetworkBenchmarkEnvironment<T extends IOReadableWritable> {
return resultPartition;
}
- private InputGate createInputGate(
- IntermediateDataSetID dataSetID,
- ExecutionAttemptID executionAttemptID,
- final TaskManagerLocation senderLocation,
- NetworkEnvironment environment,
- final int channels) throws IOException {
-
+ private InputGate createInputGate(TaskManagerLocation senderLocation) throws IOException {
InputGate[] gates = new InputGate[channels];
for (int channel = 0; channel < channels; ++channel) {
- int finalChannel = channel;
- InputChannelDeploymentDescriptor[] channelDescriptors = Arrays.stream(partitionIds)
- .map(partitionId -> new InputChannelDeploymentDescriptor(
- partitionId,
- localMode ? ResultPartitionLocation.createLocal() : ResultPartitionLocation.createRemote(new ConnectionID(senderLocation, finalChannel))))
- .toArray(InputChannelDeploymentDescriptor[]::new);
-
- final InputGateDeploymentDescriptor gateDescriptor = new InputGateDeploymentDescriptor(
- dataSetID,
- ResultPartitionType.PIPELINED_BOUNDED,
- channel,
- channelDescriptors);
-
- SingleInputGate gate = new SingleInputGateFactory(
- environment.getConfiguration(),
- environment.getConnectionManager(),
- environment.getResultPartitionManager(),
- new TaskEventDispatcher(),
- environment.getNetworkBufferPool())
- .create(
- "receiving task[" + channel + "]",
- gateDescriptor,
- SingleInputGateBuilder.NO_OP_PRODUCER_CHECKER,
- InputChannelTestUtils.newUnregisteredInputChannelMetrics(),
- new SimpleCounter());
+ final InputGateDeploymentDescriptor gateDescriptor = createInputGateDeploymentDescriptor(
+ senderLocation,
+ channel);
+
+ final SingleInputGate gate = gateFactory.create(
+ "receiving task[" + channel + "]",
+ gateDescriptor,
+ SingleInputGateBuilder.NO_OP_PRODUCER_CHECKER,
+ InputChannelTestUtils.newUnregisteredInputChannelMetrics(),
+ new SimpleCounter());
gate.setup();
gates[channel] = gate;
@@ -273,4 +254,22 @@ public class StreamNetworkBenchmarkEnvironment<T extends IOReadableWritable> {
return gates[0];
}
}
+
+ private InputGateDeploymentDescriptor createInputGateDeploymentDescriptor(
+ TaskManagerLocation senderLocation,
+ int consumedSubpartitionIndex) {
+
+ final InputChannelDeploymentDescriptor[] channelDescriptors = Arrays.stream(partitionIds)
+ .map(partitionId -> new InputChannelDeploymentDescriptor(
+ partitionId,
+ localMode ? ResultPartitionLocation.createLocal() : ResultPartitionLocation.createRemote(
+ new ConnectionID(senderLocation, consumedSubpartitionIndex))))
+ .toArray(InputChannelDeploymentDescriptor[]::new);
+
+ return new InputGateDeploymentDescriptor(
+ dataSetID,
+ ResultPartitionType.PIPELINED_BOUNDED,
+ consumedSubpartitionIndex,
+ channelDescriptors);
+ }
}
[flink] 02/03: [FLINK-12201][network,
metrics] Introduce InputGateWithMetrics in Task to increment
numBytesIn metric
Posted by pn...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
pnowojski pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 3351ca11a366713328a6d0cb0e87100cb8c24200
Author: Zhijiang <wa...@aliyun.com>
AuthorDate: Thu May 30 23:15:27 2019 +0800
[FLINK-12201][network,metrics] Introduce InputGateWithMetrics in Task to increment numBytesIn metric
Incrementing of numBytesIn metric in SingleInputGate does not depend on shuffle service and can be moved out of network
internals into Task. Task could wrap InputGate provided by ShuffleService with InputGateWithMetrics which would increment
numBytesIn metric.
---
.../runtime/io/network/NetworkEnvironment.java | 7 +-
.../network/partition/consumer/BufferOrEvent.java | 21 +++-
.../partition/consumer/SingleInputGate.java | 9 +-
.../partition/consumer/SingleInputGateFactory.java | 5 +-
.../runtime/taskmanager/InputGateWithMetrics.java | 107 +++++++++++++++++++++
.../org/apache/flink/runtime/taskmanager/Task.java | 14 ++-
.../network/partition/InputGateFairnessTest.java | 2 -
.../partition/consumer/SingleInputGateBuilder.java | 5 -
.../partition/consumer/SingleInputGateTest.java | 7 +-
.../flink/streaming/runtime/io/BufferSpiller.java | 4 +-
.../StreamNetworkBenchmarkEnvironment.java | 22 +++--
11 files changed, 156 insertions(+), 47 deletions(-)
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/NetworkEnvironment.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/NetworkEnvironment.java
index b90e903..d381203 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/NetworkEnvironment.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/NetworkEnvironment.java
@@ -20,7 +20,6 @@ package org.apache.flink.runtime.io.network;
import org.apache.flink.annotation.VisibleForTesting;
import org.apache.flink.api.common.JobID;
-import org.apache.flink.metrics.Counter;
import org.apache.flink.metrics.Gauge;
import org.apache.flink.metrics.MetricGroup;
import org.apache.flink.runtime.deployment.InputGateDeploymentDescriptor;
@@ -248,8 +247,7 @@ public class NetworkEnvironment {
Collection<InputGateDeploymentDescriptor> inputGateDeploymentDescriptors,
MetricGroup parentGroup,
MetricGroup inputGroup,
- MetricGroup buffersGroup,
- Counter numBytesInCounter) {
+ MetricGroup buffersGroup) {
synchronized (lock) {
Preconditions.checkState(!isShutdown, "The NetworkEnvironment has already been shut down.");
@@ -261,8 +259,7 @@ public class NetworkEnvironment {
taskName,
igdd,
partitionProducerStateProvider,
- inputChannelMetrics,
- numBytesInCounter);
+ inputChannelMetrics);
InputGateID id = new InputGateID(igdd.getConsumedResultId(), executionId);
inputGatesById.put(id, inputGate);
inputGate.getCloseFuture().thenRun(() -> inputGatesById.remove(id));
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/BufferOrEvent.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/BufferOrEvent.java
index d1da438..f3ba122 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/BufferOrEvent.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/BufferOrEvent.java
@@ -18,6 +18,7 @@
package org.apache.flink.runtime.io.network.partition.consumer;
+import org.apache.flink.annotation.VisibleForTesting;
import org.apache.flink.runtime.event.AbstractEvent;
import org.apache.flink.runtime.io.network.buffer.Buffer;
@@ -43,26 +44,32 @@ public class BufferOrEvent {
private int channelIndex;
- BufferOrEvent(Buffer buffer, int channelIndex, boolean moreAvailable) {
+ private final int size;
+
+ public BufferOrEvent(Buffer buffer, int channelIndex, boolean moreAvailable) {
this.buffer = checkNotNull(buffer);
this.event = null;
this.channelIndex = channelIndex;
this.moreAvailable = moreAvailable;
+ this.size = buffer.getSize();
}
- BufferOrEvent(AbstractEvent event, int channelIndex, boolean moreAvailable) {
+ public BufferOrEvent(AbstractEvent event, int channelIndex, boolean moreAvailable, int size) {
this.buffer = null;
this.event = checkNotNull(event);
this.channelIndex = channelIndex;
this.moreAvailable = moreAvailable;
+ this.size = size;
}
+ @VisibleForTesting
public BufferOrEvent(Buffer buffer, int channelIndex) {
this(buffer, channelIndex, true);
}
+ @VisibleForTesting
public BufferOrEvent(AbstractEvent event, int channelIndex) {
- this(event, channelIndex, true);
+ this(event, channelIndex, true, 0);
}
public boolean isBuffer() {
@@ -96,11 +103,15 @@ public class BufferOrEvent {
@Override
public String toString() {
- return String.format("BufferOrEvent [%s, channelIndex = %d]",
- isBuffer() ? buffer : event, channelIndex);
+ return String.format("BufferOrEvent [%s, channelIndex = %d, size = %d]",
+ isBuffer() ? buffer : event, channelIndex, size);
}
public void setMoreAvailable(boolean moreAvailable) {
this.moreAvailable = moreAvailable;
}
+
+ public int getSize() {
+ return size;
+ }
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/SingleInputGate.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/SingleInputGate.java
index 5e5a722..4e718d7 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/SingleInputGate.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/SingleInputGate.java
@@ -19,7 +19,6 @@
package org.apache.flink.runtime.io.network.partition.consumer;
import org.apache.flink.annotation.VisibleForTesting;
-import org.apache.flink.metrics.Counter;
import org.apache.flink.runtime.deployment.InputChannelDeploymentDescriptor;
import org.apache.flink.runtime.deployment.ResultPartitionLocation;
import org.apache.flink.runtime.event.AbstractEvent;
@@ -165,8 +164,6 @@ public class SingleInputGate extends InputGate {
/** A timer to retrigger local partition requests. Only initialized if actually needed. */
private Timer retriggerLocalRequestTimer;
- private final Counter numBytesIn;
-
private final SupplierWithException<BufferPool, IOException> bufferPoolFactory;
private final CompletableFuture<Void> closeFuture;
@@ -178,7 +175,6 @@ public class SingleInputGate extends InputGate {
int consumedSubpartitionIndex,
int numberOfInputChannels,
PartitionProducerStateProvider partitionProducerStateProvider,
- Counter numBytesIn,
boolean isCreditBased,
SupplierWithException<BufferPool, IOException> bufferPoolFactory) {
@@ -200,8 +196,6 @@ public class SingleInputGate extends InputGate {
this.partitionProducerStateProvider = checkNotNull(partitionProducerStateProvider);
- this.numBytesIn = checkNotNull(numBytesIn);
-
this.isCreditBased = isCreditBased;
this.closeFuture = new CompletableFuture<>();
@@ -566,7 +560,6 @@ public class SingleInputGate extends InputGate {
Buffer buffer,
boolean moreAvailable,
InputChannel currentChannel) throws IOException, InterruptedException {
- numBytesIn.inc(buffer.getSizeUnsafe());
if (buffer.isBuffer()) {
return new BufferOrEvent(buffer, currentChannel.getChannelIndex(), moreAvailable);
}
@@ -596,7 +589,7 @@ public class SingleInputGate extends InputGate {
currentChannel.releaseAllResources();
}
- return new BufferOrEvent(event, currentChannel.getChannelIndex(), moreAvailable);
+ return new BufferOrEvent(event, currentChannel.getChannelIndex(), moreAvailable, buffer.getSize());
}
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/SingleInputGateFactory.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/SingleInputGateFactory.java
index fcc3665..6caf017 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/SingleInputGateFactory.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/SingleInputGateFactory.java
@@ -19,7 +19,6 @@
package org.apache.flink.runtime.io.network.partition.consumer;
import org.apache.flink.annotation.VisibleForTesting;
-import org.apache.flink.metrics.Counter;
import org.apache.flink.runtime.deployment.InputChannelDeploymentDescriptor;
import org.apache.flink.runtime.deployment.InputGateDeploymentDescriptor;
import org.apache.flink.runtime.deployment.ResultPartitionLocation;
@@ -99,8 +98,7 @@ public class SingleInputGateFactory {
@Nonnull String owningTaskName,
@Nonnull InputGateDeploymentDescriptor igdd,
@Nonnull PartitionProducerStateProvider partitionProducerStateProvider,
- @Nonnull InputChannelMetrics metrics,
- @Nonnull Counter numBytesInCounter) {
+ @Nonnull InputChannelMetrics metrics) {
final IntermediateDataSetID consumedResultId = checkNotNull(igdd.getConsumedResultId());
final ResultPartitionType consumedPartitionType = checkNotNull(igdd.getConsumedPartitionType());
@@ -116,7 +114,6 @@ public class SingleInputGateFactory {
consumedSubpartitionIndex,
icdd.length,
partitionProducerStateProvider,
- numBytesInCounter,
isCreditBased,
createBufferPoolFactory(icdd.length, consumedPartitionType));
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/InputGateWithMetrics.java b/flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/InputGateWithMetrics.java
new file mode 100644
index 0000000..e9e3038
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/InputGateWithMetrics.java
@@ -0,0 +1,107 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.flink.runtime.taskmanager;
+
+import org.apache.flink.metrics.Counter;
+import org.apache.flink.runtime.event.TaskEvent;
+import org.apache.flink.runtime.io.network.partition.consumer.BufferOrEvent;
+import org.apache.flink.runtime.io.network.partition.consumer.InputGate;
+import org.apache.flink.runtime.metrics.groups.TaskIOMetricGroup;
+
+import java.io.IOException;
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
+/**
+ * This class wraps {@link InputGate} provided by shuffle service and it is mainly
+ * used for increasing general input metrics from {@link TaskIOMetricGroup}.
+ */
+public class InputGateWithMetrics extends InputGate {
+
+ private final InputGate inputGate;
+
+ private final Counter numBytesIn;
+
+ public InputGateWithMetrics(InputGate inputGate, Counter numBytesIn) {
+ this.inputGate = checkNotNull(inputGate);
+ this.numBytesIn = checkNotNull(numBytesIn);
+ }
+
+ @Override
+ public CompletableFuture<?> isAvailable() {
+ return inputGate.isAvailable();
+ }
+
+ @Override
+ public int getNumberOfInputChannels() {
+ return inputGate.getNumberOfInputChannels();
+ }
+
+ @Override
+ public String getOwningTaskName() {
+ return inputGate.getOwningTaskName();
+ }
+
+ @Override
+ public boolean isFinished() {
+ return inputGate.isFinished();
+ }
+
+ @Override
+ public void setup() throws IOException {
+ inputGate.setup();
+ }
+
+ @Override
+ public void requestPartitions() throws IOException, InterruptedException {
+ inputGate.requestPartitions();
+ }
+
+ @Override
+ public Optional<BufferOrEvent> getNextBufferOrEvent() throws IOException, InterruptedException {
+ return updateMetrics(inputGate.getNextBufferOrEvent());
+ }
+
+ @Override
+ public Optional<BufferOrEvent> pollNextBufferOrEvent() throws IOException, InterruptedException {
+ return updateMetrics(inputGate.pollNextBufferOrEvent());
+ }
+
+ @Override
+ public void sendTaskEvent(TaskEvent event) throws IOException {
+ inputGate.sendTaskEvent(event);
+ }
+
+ @Override
+ public int getPageSize() {
+ return inputGate.getPageSize();
+ }
+
+ @Override
+ public void close() throws Exception {
+ inputGate.close();
+ }
+
+ private Optional<BufferOrEvent> updateMetrics(Optional<BufferOrEvent> bufferOrEvent) {
+ bufferOrEvent.ifPresent(b -> numBytesIn.inc(b.getSize()));
+ return bufferOrEvent;
+ }
+}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/Task.java b/flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/Task.java
index 57d8c6b..02ef419 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/Task.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/Task.java
@@ -56,7 +56,6 @@ import org.apache.flink.runtime.io.network.partition.PartitionProducerStateProvi
import org.apache.flink.runtime.io.network.partition.ResultPartitionConsumableNotifier;
import org.apache.flink.runtime.io.network.partition.ResultPartitionID;
import org.apache.flink.runtime.io.network.partition.consumer.InputGate;
-import org.apache.flink.runtime.io.network.partition.consumer.SingleInputGate;
import org.apache.flink.runtime.jobgraph.IntermediateDataSetID;
import org.apache.flink.runtime.jobgraph.JobVertexID;
import org.apache.flink.runtime.jobgraph.tasks.AbstractInvokable;
@@ -192,7 +191,7 @@ public class Task implements Runnable, TaskActions, PartitionProducerStateProvid
private final ResultPartitionWriter[] producedPartitions;
- private final SingleInputGate[] inputGates;
+ private final InputGate[] inputGates;
/** Connection to the task manager. */
private final TaskManagerActions taskManagerActions;
@@ -380,15 +379,20 @@ public class Task implements Runnable, TaskActions, PartitionProducerStateProvid
buffersGroup);
// consumed intermediate result partitions
- this.inputGates = networkEnvironment.createInputGates(
+ InputGate[] gates = networkEnvironment.createInputGates(
taskNameWithSubtaskAndId,
executionId,
this,
inputGateDeploymentDescriptors,
metrics.getIOMetricGroup(),
inputGroup,
- buffersGroup,
- metrics.getIOMetricGroup().getNumBytesInCounter());
+ buffersGroup);
+
+ this.inputGates = new InputGate[gates.length];
+ int counter = 0;
+ for (InputGate gate : gates) {
+ inputGates[counter++] = new InputGateWithMetrics(gate, metrics.getIOMetricGroup().getNumBytesInCounter());
+ }
invokableHasBeenCanceled = new AtomicBoolean(false);
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/InputGateFairnessTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/InputGateFairnessTest.java
index d670d01..10221c5 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/InputGateFairnessTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/InputGateFairnessTest.java
@@ -18,7 +18,6 @@
package org.apache.flink.runtime.io.network.partition;
-import org.apache.flink.metrics.SimpleCounter;
import org.apache.flink.runtime.io.network.ConnectionManager;
import org.apache.flink.runtime.io.network.api.EndOfPartitionEvent;
import org.apache.flink.runtime.io.network.api.serialization.EventSerializer;
@@ -331,7 +330,6 @@ public class InputGateFairnessTest {
consumedSubpartitionIndex,
numberOfInputChannels,
SingleInputGateBuilder.NO_OP_PRODUCER_CHECKER,
- new SimpleCounter(),
isCreditBased,
STUB_BUFFER_POOL_FACTORY);
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/SingleInputGateBuilder.java b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/SingleInputGateBuilder.java
index 51eba30..8c04d5f 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/SingleInputGateBuilder.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/SingleInputGateBuilder.java
@@ -18,8 +18,6 @@
package org.apache.flink.runtime.io.network.partition.consumer;
-import org.apache.flink.metrics.Counter;
-import org.apache.flink.metrics.SimpleCounter;
import org.apache.flink.runtime.io.network.NetworkEnvironment;
import org.apache.flink.runtime.io.network.buffer.BufferPool;
import org.apache.flink.runtime.io.network.partition.PartitionProducerStateProvider;
@@ -51,8 +49,6 @@ public class SingleInputGateBuilder {
private final PartitionProducerStateProvider partitionProducerStateProvider = NO_OP_PRODUCER_CHECKER;
- private final Counter numBytesInCounter = new SimpleCounter();
-
private boolean isCreditBased = true;
private SupplierWithException<BufferPool, IOException> bufferPoolFactory = () -> {
@@ -99,7 +95,6 @@ public class SingleInputGateBuilder {
consumedSubpartitionIndex,
numberOfChannels,
partitionProducerStateProvider,
- numBytesInCounter,
isCreditBased,
bufferPoolFactory);
}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/SingleInputGateTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/SingleInputGateTest.java
index c87f99f..5bbb26a 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/SingleInputGateTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/SingleInputGateTest.java
@@ -19,7 +19,6 @@
package org.apache.flink.runtime.io.network.partition.consumer;
import org.apache.flink.core.memory.MemorySegmentFactory;
-import org.apache.flink.metrics.SimpleCounter;
import org.apache.flink.metrics.groups.UnregisteredMetricsGroup;
import org.apache.flink.runtime.deployment.InputChannelDeploymentDescriptor;
import org.apache.flink.runtime.deployment.InputGateDeploymentDescriptor;
@@ -347,8 +346,7 @@ public class SingleInputGateTest extends InputGateTestBase {
"TestTask",
gateDesc,
SingleInputGateBuilder.NO_OP_PRODUCER_CHECKER,
- InputChannelTestUtils.newUnregisteredInputChannelMetrics(),
- new SimpleCounter());
+ InputChannelTestUtils.newUnregisteredInputChannelMetrics());
try {
assertEquals(gateDesc.getConsumedPartitionType(), gate.getConsumedPartitionType());
@@ -599,8 +597,7 @@ public class SingleInputGateTest extends InputGateTestBase {
Arrays.asList(gateDescs),
new UnregisteredMetricsGroup(),
new UnregisteredMetricsGroup(),
- new UnregisteredMetricsGroup(),
- new SimpleCounter());
+ new UnregisteredMetricsGroup());
Map<InputGateID, SingleInputGate> inputGatesById = new HashMap<>();
for (int i = 0; i < numberOfGates; i++) {
inputGatesById.put(new InputGateID(ids[i], consumerID), gates[i]);
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/BufferSpiller.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/BufferSpiller.java
index 7000430..5a7c496 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/BufferSpiller.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/BufferSpiller.java
@@ -374,7 +374,7 @@ public class BufferSpiller implements BufferBlocker {
Buffer buf = new NetworkBuffer(seg, FreeingBufferRecycler.INSTANCE);
buf.setSize(length);
- return new BufferOrEvent(buf, channel);
+ return new BufferOrEvent(buf, channel, true);
}
else {
// deserialize event
@@ -399,7 +399,7 @@ public class BufferSpiller implements BufferBlocker {
AbstractEvent evt = EventSerializer.fromSerializedEvent(buffer, getClass().getClassLoader());
buffer.limit(oldLimit);
- return new BufferOrEvent(evt, channel);
+ return new BufferOrEvent(evt, channel, true, length);
}
}
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/io/benchmark/StreamNetworkBenchmarkEnvironment.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/io/benchmark/StreamNetworkBenchmarkEnvironment.java
index f00a673..7a9c863 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/io/benchmark/StreamNetworkBenchmarkEnvironment.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/io/benchmark/StreamNetworkBenchmarkEnvironment.java
@@ -47,6 +47,7 @@ import org.apache.flink.runtime.io.network.partition.consumer.SingleInputGateBui
import org.apache.flink.runtime.io.network.partition.consumer.SingleInputGateFactory;
import org.apache.flink.runtime.io.network.partition.consumer.UnionInputGate;
import org.apache.flink.runtime.jobgraph.IntermediateDataSetID;
+import org.apache.flink.runtime.taskmanager.InputGateWithMetrics;
import org.apache.flink.runtime.taskmanager.NetworkEnvironmentConfiguration;
import org.apache.flink.runtime.taskmanager.TaskManagerLocation;
import org.apache.flink.runtime.util.ConfigurationParserUtils;
@@ -237,12 +238,7 @@ public class StreamNetworkBenchmarkEnvironment<T extends IOReadableWritable> {
senderLocation,
channel);
- final SingleInputGate gate = gateFactory.create(
- "receiving task[" + channel + "]",
- gateDescriptor,
- SingleInputGateBuilder.NO_OP_PRODUCER_CHECKER,
- InputChannelTestUtils.newUnregisteredInputChannelMetrics(),
- new SimpleCounter());
+ final InputGate gate = createInputGateWithMetrics(gateFactory, gateDescriptor, channel);
gate.setup();
gates[channel] = gate;
@@ -272,4 +268,18 @@ public class StreamNetworkBenchmarkEnvironment<T extends IOReadableWritable> {
consumedSubpartitionIndex,
channelDescriptors);
}
+
+ private InputGate createInputGateWithMetrics(
+ SingleInputGateFactory gateFactory,
+ InputGateDeploymentDescriptor gateDescriptor,
+ int channelIndex) {
+
+ final SingleInputGate singleGate = gateFactory.create(
+ "receiving task[" + channelIndex + "]",
+ gateDescriptor,
+ SingleInputGateBuilder.NO_OP_PRODUCER_CHECKER,
+ InputChannelTestUtils.newUnregisteredInputChannelMetrics());
+
+ return new InputGateWithMetrics(singleGate, new SimpleCounter());
+ }
}
[flink] 03/03: [hotfix][network] Drop redundant getSizeUnsafe from
Buffer interface
Posted by pn...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
pnowojski pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit a36ec7fcecf909a1d6a8c2b85a125d0ccc8b0934
Author: Zhijiang <wa...@aliyun.com>
AuthorDate: Fri May 31 16:47:38 2019 +0800
[hotfix][network] Drop redundant getSizeUnsafe from Buffer interface
The implementations of getSize and getSizeUnsafe are exactly the same now, which do not need the synchronized way.
So We could remove the getSizeUnsafe to make it clean and clear.
---
.../org/apache/flink/runtime/io/network/buffer/Buffer.java | 11 -----------
.../apache/flink/runtime/io/network/buffer/NetworkBuffer.java | 5 -----
.../io/network/buffer/ReadOnlySlicedNetworkBuffer.java | 5 -----
.../io/network/partition/consumer/LocalInputChannel.java | 2 +-
.../io/network/partition/consumer/RemoteInputChannel.java | 2 +-
.../flink/runtime/io/network/buffer/NetworkBufferTest.java | 6 ------
.../runtime/io/network/buffer/ReadOnlySlicedBufferTest.java | 10 ++--------
7 files changed, 4 insertions(+), 37 deletions(-)
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/Buffer.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/Buffer.java
index 96b18ee..8e5d85f 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/Buffer.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/Buffer.java
@@ -159,17 +159,6 @@ public interface Buffer {
void setReaderIndex(int readerIndex) throws IndexOutOfBoundsException;
/**
- * Returns the size of the written data, i.e. the <tt>writer index</tt>, of this buffer in an
- * non-synchronized fashion.
- *
- * <p>This is where writable bytes start in the backing memory segment.
- *
- * @return writer index (from 0 (inclusive) to the size of the backing {@link MemorySegment}
- * (inclusive))
- */
- int getSizeUnsafe();
-
- /**
* Returns the size of the written data, i.e. the <tt>writer index</tt>, of this buffer.
*
* <p>This is where writable bytes start in the backing memory segment.
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/NetworkBuffer.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/NetworkBuffer.java
index 05b7582..56afbb8 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/NetworkBuffer.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/NetworkBuffer.java
@@ -291,11 +291,6 @@ public class NetworkBuffer extends AbstractReferenceCountedByteBuf implements Bu
}
@Override
- public int getSizeUnsafe() {
- return writerIndex();
- }
-
- @Override
public int getSize() {
return writerIndex();
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/ReadOnlySlicedNetworkBuffer.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/ReadOnlySlicedNetworkBuffer.java
index 00e1154..dd421b7 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/ReadOnlySlicedNetworkBuffer.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/ReadOnlySlicedNetworkBuffer.java
@@ -153,11 +153,6 @@ public final class ReadOnlySlicedNetworkBuffer extends ReadOnlyByteBuf implement
}
@Override
- public int getSizeUnsafe() {
- return writerIndex();
- }
-
- @Override
public int getSize() {
return writerIndex();
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/LocalInputChannel.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/LocalInputChannel.java
index feb6b72..3a54310 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/LocalInputChannel.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/LocalInputChannel.java
@@ -193,7 +193,7 @@ public class LocalInputChannel extends InputChannel implements BufferAvailabilit
}
}
- numBytesIn.inc(next.buffer().getSizeUnsafe());
+ numBytesIn.inc(next.buffer().getSize());
numBuffersIn.inc();
return Optional.of(new BufferAndAvailability(next.buffer(), next.isMoreAvailable(), next.buffersInBacklog()));
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/RemoteInputChannel.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/RemoteInputChannel.java
index 50bf1d0..a42a93b 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/RemoteInputChannel.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/RemoteInputChannel.java
@@ -201,7 +201,7 @@ public class RemoteInputChannel extends InputChannel implements BufferRecycler,
moreAvailable = !receivedBuffers.isEmpty();
}
- numBytesIn.inc(next.getSizeUnsafe());
+ numBytesIn.inc(next.getSize());
numBuffersIn.inc();
return Optional.of(new BufferAndAvailability(next, moreAvailable, getSenderBacklog()));
}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/buffer/NetworkBufferTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/buffer/NetworkBufferTest.java
index 47615d9..1cc4fc3 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/buffer/NetworkBufferTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/buffer/NetworkBufferTest.java
@@ -216,7 +216,6 @@ public class NetworkBufferTest extends AbstractByteBufTest {
assertEquals(0, slice.getReaderIndex());
assertEquals(10, slice.getSize());
- assertEquals(10, slice.getSizeUnsafe());
assertSame(buffer, slice.unwrap().unwrap());
// slice indices should be independent:
@@ -224,7 +223,6 @@ public class NetworkBufferTest extends AbstractByteBufTest {
buffer.setReaderIndex(2);
assertEquals(0, slice.getReaderIndex());
assertEquals(10, slice.getSize());
- assertEquals(10, slice.getSizeUnsafe());
}
@Test
@@ -244,7 +242,6 @@ public class NetworkBufferTest extends AbstractByteBufTest {
assertEquals(0, slice.getReaderIndex());
assertEquals(10, slice.getSize());
- assertEquals(10, slice.getSizeUnsafe());
assertSame(buffer, slice.unwrap().unwrap());
// slice indices should be independent:
@@ -252,7 +249,6 @@ public class NetworkBufferTest extends AbstractByteBufTest {
buffer.setReaderIndex(2);
assertEquals(0, slice.getReaderIndex());
assertEquals(10, slice.getSize());
- assertEquals(10, slice.getSizeUnsafe());
}
@Test
@@ -312,13 +308,11 @@ public class NetworkBufferTest extends AbstractByteBufTest {
NetworkBuffer buffer = newBuffer(1024, 1024, isBuffer);
assertEquals(0, buffer.getSize()); // initially 0
- assertEquals(0, buffer.getSizeUnsafe());
assertEquals(buffer.writerIndex(), buffer.getSize());
assertEquals(0, buffer.readerIndex()); // initially 0
buffer.setSize(10);
assertEquals(10, buffer.getSize());
- assertEquals(10, buffer.getSizeUnsafe());
assertEquals(buffer.writerIndex(), buffer.getSize());
assertEquals(0, buffer.readerIndex()); // independent
}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/buffer/ReadOnlySlicedBufferTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/buffer/ReadOnlySlicedBufferTest.java
index d5814ee..444fdc7 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/buffer/ReadOnlySlicedBufferTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/buffer/ReadOnlySlicedBufferTest.java
@@ -233,8 +233,7 @@ public class ReadOnlySlicedBufferTest {
/**
* Tests the independence of the writer index via
* {@link ReadOnlySlicedNetworkBuffer#setSize(int)},
- * {@link ReadOnlySlicedNetworkBuffer#getSize()}, and
- * {@link ReadOnlySlicedNetworkBuffer#getSizeUnsafe()}.
+ * {@link ReadOnlySlicedNetworkBuffer#getSize()}.
*/
@Test
public void testGetSetSize1() {
@@ -244,8 +243,7 @@ public class ReadOnlySlicedBufferTest {
/**
* Tests the independence of the writer index via
* {@link ReadOnlySlicedNetworkBuffer#setSize(int)},
- * {@link ReadOnlySlicedNetworkBuffer#getSize()}, and
- * {@link ReadOnlySlicedNetworkBuffer#getSizeUnsafe()}.
+ * {@link ReadOnlySlicedNetworkBuffer#getSize()}.
*/
@Test
public void testGetSetSize2() {
@@ -254,14 +252,10 @@ public class ReadOnlySlicedBufferTest {
private void testGetSetSize(ReadOnlySlicedNetworkBuffer slice, int sliceSize) {
assertEquals(DATA_SIZE, buffer.getSize());
- assertEquals(DATA_SIZE, buffer.getSizeUnsafe());
assertEquals(sliceSize, slice.getSize());
- assertEquals(sliceSize, slice.getSizeUnsafe());
buffer.setSize(DATA_SIZE + 1);
assertEquals(DATA_SIZE + 1, buffer.getSize());
- assertEquals(DATA_SIZE + 1, buffer.getSizeUnsafe());
assertEquals(sliceSize, slice.getSize());
- assertEquals(sliceSize, slice.getSizeUnsafe());
}
@Test