[GitHub] flink pull request: [FLINK-1478] Add support for strictly local in...
Github user rmetzger commented on the pull request: https://github.com/apache/flink/pull/375#issuecomment-73477321 Looks good. --- If your project is set up for it, you can reply to this email and have your reply appear on GitHub as well. If your project does not have this feature enabled and wishes so, or if the feature is enabled but not working, please contact infrastructure at infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] flink pull request: [FLINK-1478] Add support for strictly local in...
Github user fhueske commented on the pull request: https://github.com/apache/flink/pull/375#issuecomment-73505358 Only minor remarks. Looks good otherwise. --- If your project is set up for it, you can reply to this email and have your reply appear on GitHub as well. If your project does not have this feature enabled and wishes so, or if the feature is enabled but not working, please contact infrastructure at infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] flink pull request: [FLINK-1478] Add support for strictly local in...
Github user fhueske commented on a diff in the pull request:
https://github.com/apache/flink/pull/375#discussion_r24327047
--- Diff:
flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ExecutionJobVertex.java
---
@@ -260,15 +260,49 @@ public void
connectToPredecessors(Map
public void scheduleAll(Scheduler scheduler, boolean queued) throws
NoResourceAvailableException {
-// ExecutionVertex[] vertices = this.taskVertices;
-//
-// for (int i = 0; i < vertices.length; i++) {
-// ExecutionVertex v = vertices[i];
-//
-// if (v.get
-// }
+ ExecutionVertex[] vertices = this.taskVertices;
- for (ExecutionVertex ev : getTaskVertices()) {
+ // check if we need to do pre-assignment of tasks
+ if (inputSplitsPerSubtask != null) {
+
+ final Map> instances =
scheduler.getInstancesByHost();
+ final Map assignments = new
HashMap();
+
+ for (int i = 0; i < vertices.length; i++) {
+ List splitsForHost =
inputSplitsPerSubtask[i];
+ if (splitsForHost == null ||
splitsForHost.isEmpty()) {
+ continue;
+ }
+
+ String[] hostNames =
splitsForHost.get(0).getHostnames();
+ if (hostNames == null || hostNames.length == 0
|| hostNames[0] == null) {
+ continue;
+ }
+
+ String host = hostNames[0];
+ ExecutionVertex v = vertices[i];
+
+ List instancesOnHost =
instances.get(host);
+
+ if (instancesOnHost == null ||
instancesOnHost.isEmpty()) {
+ throw new
NoResourceAvailableException("Cannot schedule a strictly local task to host " +
host
+ + ". No TaskManager
available on that host.");
+ }
+
+ Integer pos = assignments.get(host);
+ if (pos == null) {
+ pos = 0;
+ assignments.put(host, 0);
+ } else {
+ assignments.put(host, pos + 1 %
instancesOnHost.size());
--- End diff --
Doesn't this potentially cause multiple subtasks being assigned to the same
instance?
I guess that would fail in the scheduler. Shouldn't we catch the case here
and return a more detailed exception why scheduling constraint could not be
fulfilled?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-1478] Add support for strictly local in...
Github user StephanEwen commented on a diff in the pull request:
https://github.com/apache/flink/pull/375#discussion_r24329907
--- Diff:
flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ExecutionJobVertex.java
---
@@ -260,15 +260,49 @@ public void
connectToPredecessors(Map
public void scheduleAll(Scheduler scheduler, boolean queued) throws
NoResourceAvailableException {
-// ExecutionVertex[] vertices = this.taskVertices;
-//
-// for (int i = 0; i < vertices.length; i++) {
-// ExecutionVertex v = vertices[i];
-//
-// if (v.get
-// }
+ ExecutionVertex[] vertices = this.taskVertices;
- for (ExecutionVertex ev : getTaskVertices()) {
+ // check if we need to do pre-assignment of tasks
+ if (inputSplitsPerSubtask != null) {
+
+ final Map> instances =
scheduler.getInstancesByHost();
+ final Map assignments = new
HashMap();
+
+ for (int i = 0; i < vertices.length; i++) {
+ List splitsForHost =
inputSplitsPerSubtask[i];
+ if (splitsForHost == null ||
splitsForHost.isEmpty()) {
+ continue;
+ }
+
+ String[] hostNames =
splitsForHost.get(0).getHostnames();
+ if (hostNames == null || hostNames.length == 0
|| hostNames[0] == null) {
+ continue;
+ }
+
+ String host = hostNames[0];
+ ExecutionVertex v = vertices[i];
+
+ List instancesOnHost =
instances.get(host);
+
+ if (instancesOnHost == null ||
instancesOnHost.isEmpty()) {
+ throw new
NoResourceAvailableException("Cannot schedule a strictly local task to host " +
host
+ + ". No TaskManager
available on that host.");
+ }
+
+ Integer pos = assignments.get(host);
+ if (pos == null) {
+ pos = 0;
+ assignments.put(host, 0);
+ } else {
+ assignments.put(host, pos + 1 %
instancesOnHost.size());
--- End diff --
It should be possible that multiple subtasks go to the same instance. If
there are too many, it would fail in the scheduler, yes. We can check the the
number of subtasks on the instance does not exceed the number of slots.
This seems to me like a workaround solution anyways (until we can tie
splits to tasks), so it might be okay.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-1478] Add support for strictly local in...
Github user fhueske commented on a diff in the pull request:
https://github.com/apache/flink/pull/375#discussion_r24330528
--- Diff:
flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ExecutionJobVertex.java
---
@@ -260,15 +260,49 @@ public void
connectToPredecessors(Map
public void scheduleAll(Scheduler scheduler, boolean queued) throws
NoResourceAvailableException {
-// ExecutionVertex[] vertices = this.taskVertices;
-//
-// for (int i = 0; i < vertices.length; i++) {
-// ExecutionVertex v = vertices[i];
-//
-// if (v.get
-// }
+ ExecutionVertex[] vertices = this.taskVertices;
- for (ExecutionVertex ev : getTaskVertices()) {
+ // check if we need to do pre-assignment of tasks
+ if (inputSplitsPerSubtask != null) {
+
+ final Map> instances =
scheduler.getInstancesByHost();
+ final Map assignments = new
HashMap();
+
+ for (int i = 0; i < vertices.length; i++) {
+ List splitsForHost =
inputSplitsPerSubtask[i];
+ if (splitsForHost == null ||
splitsForHost.isEmpty()) {
+ continue;
+ }
+
+ String[] hostNames =
splitsForHost.get(0).getHostnames();
+ if (hostNames == null || hostNames.length == 0
|| hostNames[0] == null) {
+ continue;
+ }
+
+ String host = hostNames[0];
+ ExecutionVertex v = vertices[i];
+
+ List instancesOnHost =
instances.get(host);
+
+ if (instancesOnHost == null ||
instancesOnHost.isEmpty()) {
+ throw new
NoResourceAvailableException("Cannot schedule a strictly local task to host " +
host
+ + ". No TaskManager
available on that host.");
+ }
+
+ Integer pos = assignments.get(host);
+ if (pos == null) {
+ pos = 0;
+ assignments.put(host, 0);
+ } else {
+ assignments.put(host, pos + 1 %
instancesOnHost.size());
--- End diff --
Ah, yes sure. I confused instances and slots...
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-1478] Add support for strictly local in...
Github user asfgit closed the pull request at: https://github.com/apache/flink/pull/375 --- If your project is set up for it, you can reply to this email and have your reply appear on GitHub as well. If your project does not have this feature enabled and wishes so, or if the feature is enabled but not working, please contact infrastructure at infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
