<commit_before>/*
 * Copyright 1999-2015 dangdang.com.
 * <p>
 * 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.
 * </p>
 */

package com.dangdang.ddframe.job.cloud.scheduler.mesos;

import com.dangdang.ddframe.job.cloud.scheduler.config.CloudJobConfiguration;
import com.dangdang.ddframe.job.cloud.scheduler.context.ExecutionType;
import com.dangdang.ddframe.job.cloud.scheduler.context.JobContext;
import com.dangdang.ddframe.job.cloud.scheduler.context.TaskContext;
import com.dangdang.ddframe.job.executor.ShardingContexts;
import com.dangdang.ddframe.job.util.BlockUtils;
import com.dangdang.ddframe.job.util.config.ShardingItemParameters;
import com.google.common.base.Function;
import com.google.common.collect.Lists;
import com.google.protobuf.ByteString;
import com.netflix.fenzo.TaskAssignmentResult;
import com.netflix.fenzo.TaskRequest;
import com.netflix.fenzo.TaskScheduler;
import com.netflix.fenzo.VMAssignmentResult;
import com.netflix.fenzo.VirtualMachineLease;
import lombok.RequiredArgsConstructor;
import org.apache.mesos.Protos;
import org.apache.mesos.SchedulerDriver;

import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;

/**
 * 任务启动处理器.
 *
 * @author zhangliang
 */
@RequiredArgsConstructor
public final class TaskLaunchProcessor implements Runnable {
    
    private static volatile boolean shutdown;
    
    private final LeasesQueue leasesQueue;
    
    private final SchedulerDriver schedulerDriver;
    
    private final TaskScheduler taskScheduler;
    
    private final FacadeService facadeService;
    
    /**
     * 线程关闭.
     */
    public static void shutdown() {
        shutdown = true;
    }
    
    @Override
    public void run() {
        while (!shutdown) {
            Collection<JobContext> eligibleJobContexts =  facadeService.getEligibleJobContext();
            Map<String, Integer> jobShardingTotalCountMap = new HashMap<>(eligibleJobContexts.size(), 1);
            List<TaskRequest> pendingTasks = new ArrayList<>(eligibleJobContexts.size() * 10);
            for (JobContext each : eligibleJobContexts) {
                pendingTasks.addAll(getTaskRequests(each));
                if (ExecutionType.FAILOVER != each.getType()) {
                    jobShardingTotalCountMap.put(each.getJobConfig().getJobName(), each.getJobConfig().getTypeConfig().getCoreConfig().getShardingTotalCount());
                }
            }
            Collection<VMAssignmentResult> vmAssignmentResults = taskScheduler.scheduleOnce(pendingTasks, leasesQueue.drainTo()).getResultMap().values();
            Collection<String> integrityViolationJobs = getIntegrityViolationJobs(jobShardingTotalCountMap, vmAssignmentResults);
            for (VMAssignmentResult each: vmAssignmentResults) {
                List<VirtualMachineLease> leasesUsed = each.getLeasesUsed();
                List<Protos.TaskInfo> taskInfoList = new ArrayList<>(each.getTasksAssigned().size() * 10);
                taskInfoList.addAll(getTaskInfoList(integrityViolationJobs, each, leasesUsed.get(0).hostname(), leasesUsed.get(0).getOffer().getSlaveId()));
                schedulerDriver.launchTasks(getOfferIDs(leasesUsed), taskInfoList);
                facadeService.removeLaunchTasksFromQueue(Lists.transform(taskInfoList, new Function<Protos.TaskInfo, TaskContext>() {
                    
                    @Override
                    public TaskContext apply(final Protos.TaskInfo input) {
                        return TaskContext.from(input.getTaskId().getValue());
                    }
                }));
                for (Protos.TaskInfo taskInfo : taskInfoList) {
                    facadeService.addRunning(TaskContext.from(taskInfo.getTaskId().getValue()));
                }
            }
            BlockUtils.waitingShortTime();
        }
    }
    
    private Collection<TaskRequest> getTaskRequests(final JobContext jobContext) {
        Collection<TaskRequest> result = new ArrayList<>(jobContext.getAssignedShardingItems().size());
        CloudJobConfiguration jobConfig = jobContext.getJobConfig();
        for (int each : jobContext.getAssignedShardingItems()) {
            result.add(new JobTaskRequest(new TaskContext(jobConfig.getJobName(), each, jobContext.getType(), "fake-slave"), jobConfig));
        }
        return result;
    }
    
    private Collection<String> getIntegrityViolationJobs(final Map<String, Integer> jobShardingTotalCountMap, final Collection<VMAssignmentResult> vmAssignmentResults) {
        Map<String, Integer> assignedJobShardingTotalCountMap = new HashMap<>(jobShardingTotalCountMap.size(), 1);
        for (VMAssignmentResult vmAssignmentResult: vmAssignmentResults) {
            for (TaskAssignmentResult tasksAssigned: vmAssignmentResult.getTasksAssigned()) {
                String jobName = TaskContext.from(tasksAssigned.getTaskId()).getMetaInfo().getJobName();
                if (assignedJobShardingTotalCountMap.containsKey(jobName)) {
                    assignedJobShardingTotalCountMap.put(jobName, assignedJobShardingTotalCountMap.get(jobName) + 1);
                } else {
                    assignedJobShardingTotalCountMap.put(jobName, 1);
                }
            }
        }
        Collection<String> result = new HashSet<>(assignedJobShardingTotalCountMap.size(), 1);<commit_msg>resource not enough print warn log<commit_after>/*
 * Copyright 1999-2015 dangdang.<EMAIL>.
 * <p>
 * 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.
 * </p>
 */

package com.dangdang.ddframe.job.cloud.scheduler.mesos;

import com.dangdang.ddframe.job.cloud.scheduler.config.CloudJobConfiguration;
import com.dangdang.ddframe.job.cloud.scheduler.context.ExecutionType;
import com.dangdang.ddframe.job.cloud.scheduler.context.JobContext;
import com.dangdang.ddframe.job.cloud.scheduler.context.TaskContext;
import com.dangdang.ddframe.job.executor.ShardingContexts;
import com.dangdang.ddframe.job.util.BlockUtils;
import com.dangdang.ddframe.job.util.config.ShardingItemParameters;
import com.google.common.base.Function;
import com.google.common.collect.Lists;
import com.google.protobuf.ByteString;
import com.netflix.fenzo.TaskAssignmentResult;
import com.netflix.fenzo.TaskRequest;
import com.netflix.fenzo.TaskScheduler;
import com.netflix.fenzo.VMAssignmentResult;
import com.netflix.fenzo.VirtualMachineLease;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.mesos.Protos;
import org.apache.mesos.SchedulerDriver;

import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;

/**
 * 任务启动处理器.
 *
 * @author zhangliang
 */
@RequiredArgsConstructor
@Slf4j
public final class TaskLaunchProcessor implements Runnable {
    
    private static volatile boolean shutdown;
    
    private final LeasesQueue leasesQueue;
    
    private final SchedulerDriver schedulerDriver;
    
    private final TaskScheduler taskScheduler;
    
    private final FacadeService facadeService;
    
    /**
     * 线程关闭.
     */
    public static void shutdown() {
        shutdown = true;
    }
    
    @Override
    public void run() {
        while (!shutdown) {
            Collection<JobContext> eligibleJobContexts =  facadeService.getEligibleJobContext();
            Map<String, Integer> jobShardingTotalCountMap = new HashMap<>(eligibleJobContexts.size(), 1);
            List<TaskRequest> pendingTasks = new ArrayList<>(eligibleJobContexts.size() * 10);
            for (JobContext each : eligibleJobContexts) {
                pendingTasks.addAll(getTaskRequests(each));
                if (ExecutionType.FAILOVER != each.getType()) {
                    jobShardingTotalCountMap.put(each.getJobConfig().getJobName(), each.getJobConfig().getTypeConfig().getCoreConfig().getShardingTotalCount());
                }
            }
            Collection<VMAssignmentResult> vmAssignmentResults = taskScheduler.scheduleOnce(pendingTasks, leasesQueue.drainTo()).getResultMap().values();
            logUnassignedJobs(eligibleJobContexts, vmAssignmentResults);
            Collection<String> integrityViolationJobs = getIntegrityViolationJobs(jobShardingTotalCountMap, vmAssignmentResults);
            logIntegrityViolationJobs(integrityViolationJobs);
            for (VMAssignmentResult each: vmAssignmentResults) {
                List<VirtualMachineLease> leasesUsed = each.getLeasesUsed();
                List<Protos.TaskInfo> taskInfoList = new ArrayList<>(each.getTasksAssigned().size() * 10);
                taskInfoList.addAll(getTaskInfoList(integrityViolationJobs, each, leasesUsed.get(0).hostname(), leasesUsed.get(0).getOffer().getSlaveId()));
                schedulerDriver.launchTasks(getOfferIDs(leasesUsed), taskInfoList);
                facadeService.removeLaunchTasksFromQueue(Lists.transform(taskInfoList, new Function<Protos.TaskInfo, TaskContext>() {
                    
                    @Override
                    public TaskContext apply(final Protos.TaskInfo input) {
                        return TaskContext.from(input.getTaskId().getValue());
                    }
                }));
                for (Protos.TaskInfo taskInfo : taskInfoList) {
                    facadeService.addRunning(TaskContext.from(taskInfo.getTaskId().getValue()));
                }
            }
            BlockUtils.waitingShortTime();
        }
    }
    
    private Collection<TaskRequest> getTaskRequests(final JobContext jobContext) {
        Collection<TaskRequest> result = new ArrayList<>(jobContext.getAssignedShardingItems().size());
        CloudJobConfiguration jobConfig = jobContext.getJobConfig();
        for (int each : jobContext.getAssignedShardingItems()) {
            result.add(new JobTaskRequest(new TaskContext(jobConfig.getJobName(), each, jobContext.getType(), "fake-slave"), jobConfig));
        }
        return result;
    }
    
    private void logUnassignedJobs(final Collection<JobContext> eligibleJobContexts, final Collection<VMAssignmentResult> vmAssignmentResults) {
        for (JobContext each : eligibleJobContexts) {
            if (!isAssigned(each, vmAssignmentResults)) {
                log.warn("Job {} is not assigned at this time, because resources not enough.", each.getJobConfig().getJobName());
            }
        }
    }
    
    private boolean isAssigned(final JobContext jobContext, final Collection<VMAssignmentResult> vmAssignmentResults) {
        for (VMAssignmentResult vmAssignmentResult: vmAssignmentResults) {
            for (TaskAssignmentResult taskAssignmentResult : vmAssignmentResult.getTasksAssigned()) {
                if (jobContext.getJobConfig().getJobName().equals(TaskContext.from(taskAssignmentResult.getTaskId()).getMetaInfo().getJobName())) {
                    return true;
                }
            }
        }
        return false;
    }
    
    private void logIntegrityViolationJobs(final Collection<String> integrityViolationJobs) {
        for (String each : integrityViolationJobs) {
            log.warn("Job {} is not assigned at this time, because resources not enough to run all sharding instances.", each);
        }
    }
    
    private Collection<String> getIntegrityViolationJobs(final Map<String, Integer> jobShardingTotalCountMap, final Collection<VMAssignmentResult> vmAssignmentResults) {
        Map<String, Integer> assignedJobShardingTotalCountMap = new HashMap<>(jobShardingTotalCountMap.size(), 1);
        for (VMAssignmentResult vmAssignmentResult: vmAssignmentResults) {
            for (TaskAssignmentResult tasksAssigned: vmAssignmentResult.getTasksAssigned()) {
                String jobName = TaskContext.from(tasksAssigned.getTaskId()).getMetaInfo().getJobName();
                if (assignedJobShardingTotalCountMap.containsKey(jobName)) {
                    assignedJobShardingTotalCountMap.put(jobName, assignedJobShardingTotalCountMap.get(jobName) + 1);
                } else {
                    assignedJobShardingTotalCountMap.put(jobName, 1);
                }
            }
        }
        Collection<String> result = new HashSet<>(assignedJobShardingTotalCountMap.size(), 1);
{'fraction_non_alphanumeric': 0.09476459510357815, 'fraction_numerical': 0.0036911487758945387, 'mean_word_length': 3.23882503192848, 'pattern_counts': {'":': 0, '<': 54, '<?xml version=': 0, '>': 54, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3147442', 'n_tokens_mistral': 3648, 'n_tokens_neox': 3610, 'n_words': 705}