[
https://issues.apache.org/jira/browse/FLINK-4347?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15430935#comment-15430935
]
ASF GitHub Bot commented on FLINK-4347:
---------------------------------------
Github user tillrohrmann commented on a diff in the pull request:
https://github.com/apache/flink/pull/2388#discussion_r75694327
--- Diff:
flink-runtime/src/main/java/org/apache/flink/runtime/clusterframework/SlotManager.java
---
@@ -0,0 +1,519 @@
+/*
+ * 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.clusterframework;
+
+import com.google.common.collect.BiMap;
+import com.google.common.collect.HashBiMap;
+import org.apache.flink.annotation.VisibleForTesting;
+import org.apache.flink.runtime.clusterframework.types.AllocationID;
+import org.apache.flink.runtime.clusterframework.types.ResourceID;
+import org.apache.flink.runtime.clusterframework.types.ResourceProfile;
+import org.apache.flink.runtime.clusterframework.types.ResourceSlot;
+import org.apache.flink.runtime.clusterframework.types.SlotID;
+import org.apache.flink.runtime.rpc.resourcemanager.ResourceManagerGateway;
+import org.apache.flink.runtime.rpc.resourcemanager.SlotRequest;
+import org.apache.flink.runtime.rpc.taskexecutor.SlotReport;
+import org.apache.flink.runtime.rpc.taskexecutor.SlotStatus;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.HashMap;
+import java.util.LinkedHashMap;
+import java.util.Map;
+
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
+/**
+ * SlotManager is responsible for receiving slot requests and do slot
allocations. It allows to request
+ * slots from registered TaskManagers and issues container allocation
requests in case of there are not
+ * enough available slots. Besides, it should sync its slot allocation
with TaskManager's heartbeat.
+ * <p>
+ * The main operation principle of SlotManager is:
+ * <ul>
+ * <li>1. All slot allocation status should be synced with TaskManager,
which is the ground truth.</li>
+ * <li>2. All slots that have registered must be tracked, either by free
pool or allocated pool.</li>
+ * <li>3. All slot requests will be handled by best efforts, there is no
guarantee that one request will be
+ * fulfilled in time or correctly allocated. Conflicts or timeout or some
special error will happen, it should
+ * be handled outside SlotManager. SlotManager will make each decision
based on the information it currently
+ * holds.</li>
+ * </ul>
+ * <b>IMPORTANT:</b> This class is <b>Not Thread-safe</b>.
+ */
+public abstract class SlotManager {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(SlotManager.class);
+
+ /** Gateway to communicate with ResourceManager */
+ private final ResourceManagerGateway resourceManagerGateway;
+
+ /** All registered slots, including free and allocated slots */
+ private final Map<ResourceID, Map<SlotID, ResourceSlot>>
registeredSlots;
+
+ /** All pending slot requests, waiting available slots to fulfil */
+ private final Map<AllocationID, SlotRequest> pendingSlotRequests;
+
+ /** All free slots that can be used to be allocated */
+ private final Map<SlotID, ResourceSlot> freeSlots;
+
+ /** All allocations, we can lookup allocations either by SlotID or
AllocationID */
+ private final AllocationMap allocationMap;
+
+ public SlotManager(ResourceManagerGateway resourceManagerGateway) {
+ this.resourceManagerGateway = resourceManagerGateway;
--- End diff --
`checkNotNull` would be good here
> Implement SlotManager core
> --------------------------
>
> Key: FLINK-4347
> URL: https://issues.apache.org/jira/browse/FLINK-4347
> Project: Flink
> Issue Type: Sub-task
> Components: Cluster Management
> Reporter: Kurt Young
> Assignee: Kurt Young
>
> The slot manager is responsible to maintain the list of slot requests and
> slot allocations. It allows to request slots from the registered
> TaskExecutors and issues container allocation requests in case that there are
> not enough available resources.
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
