<commit_before>/*
 * Copyright (c) 2008-2012, Hazel Bilisim Ltd. All Rights Reserved.
 *
 * 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 com.hazelcast.impl;

import com.hazelcast.config.ExecutorConfig;
import com.hazelcast.core.DistributedTask;
import com.hazelcast.core.Member;
import com.hazelcast.impl.executor.ParallelExecutor;
import com.hazelcast.impl.executor.ParallelExecutorService;
import com.hazelcast.nio.Address;
import com.hazelcast.nio.Data;
import com.hazelcast.partition.Partition;
import com.hazelcast.security.SecureCallable;
import com.hazelcast.util.Clock;

import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.Set;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicLong;
import java.util.logging.Level;

import static com.hazelcast.impl.ClusterOperation.CANCEL_EXECUTION;
import static com.hazelcast.impl.ClusterOperation.EXECUTE;
import static com.hazelcast.impl.Constants.Objects.*;
import static com.hazelcast.nio.IOUtil.toData;
import static com.hazelcast.nio.IOUtil.toObject;

public class ExecutorManager extends BaseManager {

    private final ConcurrentMap<String, NamedExecutorService> mapExecutors = new ConcurrentHashMap<String, NamedExecutorService>(10);

    private final ConcurrentMap<Thread, CallContext> mapThreadCallContexts = new ConcurrentHashMap<Thread, CallContext>(100);

    private final ParallelExecutor mapLoaderExecutorService;
    private final ParallelExecutor asyncExecutorService;
    private final NamedExecutorService defaultExecutorService;
    private final NamedExecutorService queryExecutorService;
    private final NamedExecutorService eventExecutorService;

    private volatile boolean started = false;
    private static final String DEFAULT_EXECUTOR_SERVICE = "x:default";
    private static final String QUERY_EXECUTOR_SERVICE = "x:hz.query";
    private static final String STORE_EXECUTOR_SERVICE = "x:hz.store";
    private static final String EVENT_EXECUTOR_SERVICE = "x:hz.events";
    private final Object CREATE_LOCK = new Object();
    private final ParallelExecutorService parallelExecutorService;
    private final ThreadPoolExecutor threadPoolExecutor;
    private final ConcurrentMap<ExecutionKey, RequestExecutor> executions = new ConcurrentHashMap<ExecutionKey, RequestExecutor>(100);
    private final ScheduledThreadPoolExecutor esScheduled;

    final AtomicLong executionIdGen = new AtomicLong();

    ExecutorManager(final Node node) {
        super(node);
        logger.log(Level.FINEST, "Starting ExecutorManager");
        GroupProperties gp = node.groupProperties;
        ClassLoader classLoader = node.getConfig().getClassLoader();
        threadPoolExecutor = new ThreadPoolExecutor(
                5, Integer.MAX_VALUE,
                6L,
                TimeUnit.SECONDS,
                new SynchronousQueue(),
                new ExecutorThreadFactory(node.threadGroup, node.getThreadPoolNamePrefix("cached"), classLoader),
                new RejectionHandler()) {
            protected void beforeExecute(Thread t, Runnable r) {
                threadPoolBeforeExecute(t, r);
            }
        };
        esScheduled = new ScheduledThreadPoolExecutor(2, new ExecutorThreadFactory(node.threadGroup,
                node.getThreadPoolNamePrefix("scheduled"), classLoader), new RejectionHandler()) {
            protected void beforeExecute(Thread t, Runnable r) {
                threadPoolBeforeExecute(t, r);
            }
        };
        parallelExecutorService = new ParallelExecutorService(node.getLogger(ParallelExecutorService.class.getName()), threadPoolExecutor);
        defaultExecutorService = getOrCreateNamedExecutorService(DEFAULT_EXECUTOR_SERVICE);
        queryExecutorService = getOrCreateNamedExecutorService(QUERY_EXECUTOR_SERVICE, gp.EXECUTOR_QUERY_THREAD_COUNT);
        eventExecutorService = getOrCreateNamedExecutorService(EVENT_EXECUTOR_SERVICE, gp.EXECUTOR_EVENT_THREAD_COUNT);
        mapLoaderExecutorService = parallelExecutorService.newParallelExecutor(gp.MAP_LOAD_THREAD_COUNT.getInteger());
        asyncExecutorService = parallelExecutorService.newBlockingParallelExecutor(24, 1000);
        registerPacketProcessor(EXECUTE, new ExecutionOperationHandler());
        registerPacketProcessor(CANCEL_EXECUTION, new ExecutionCancelOperationHandler());
        started = true;
    }
<commit_msg>ThreadContext memory leak fix.<commit_after>/*
 * Copyright (c) 2008-2012, Hazel Bilisim Ltd. All Rights Reserved.
 *
 * 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 com.hazelcast.impl;

import com.hazelcast.config.ExecutorConfig;
import com.hazelcast.core.DistributedTask;
import com.hazelcast.core.Member;
import com.hazelcast.impl.executor.ParallelExecutor;
import com.hazelcast.impl.executor.ParallelExecutorService;
import com.hazelcast.nio.Address;
import com.hazelcast.nio.Data;
import com.hazelcast.partition.Partition;
import com.hazelcast.security.SecureCallable;
import com.hazelcast.util.Clock;

import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.Set;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicLong;
import java.util.logging.Level;

import static com.hazelcast.impl.ClusterOperation.CANCEL_EXECUTION;
import static com.hazelcast.impl.ClusterOperation.EXECUTE;
import static com.hazelcast.impl.Constants.Objects.*;
import static com.hazelcast.nio.IOUtil.toData;
import static com.hazelcast.nio.IOUtil.toObject;

public class ExecutorManager extends BaseManager {

    private final ConcurrentMap<String, NamedExecutorService> mapExecutors = new ConcurrentHashMap<String, NamedExecutorService>(10);

    private final ConcurrentMap<Thread, CallContext> mapThreadCallContexts = new ConcurrentHashMap<Thread, CallContext>(100);

    private final ParallelExecutor mapLoaderExecutorService;
    private final ParallelExecutor asyncExecutorService;
    private final NamedExecutorService defaultExecutorService;
    private final NamedExecutorService queryExecutorService;
    private final NamedExecutorService eventExecutorService;

    private volatile boolean started = false;
    private static final String DEFAULT_EXECUTOR_SERVICE = "x:default";
    private static final String QUERY_EXECUTOR_SERVICE = "x:hz.query";
    private static final String STORE_EXECUTOR_SERVICE = "x:hz.store";
    private static final String EVENT_EXECUTOR_SERVICE = "x:hz.events";
    private final Object CREATE_LOCK = new Object();
    private final ParallelExecutorService parallelExecutorService;
    private final ThreadPoolExecutor threadPoolExecutor;
    private final ConcurrentMap<ExecutionKey, RequestExecutor> executions = new ConcurrentHashMap<ExecutionKey, RequestExecutor>(100);
    private final ScheduledThreadPoolExecutor esScheduled;

    final AtomicLong executionIdGen = new AtomicLong();

    ExecutorManager(final Node node) {
        super(node);
        logger.log(Level.FINEST, "Starting ExecutorManager");
        GroupProperties gp = node.groupProperties;
        ClassLoader classLoader = node.getConfig().getClassLoader();
        threadPoolExecutor = new ThreadPoolExecutor(
                5, Integer.MAX_VALUE,
                60L,
                TimeUnit.SECONDS,
                new SynchronousQueue(),
                new ExecutorThreadFactory(node.threadGroup, node.getThreadPoolNamePrefix("cached"), classLoader),
                new RejectionHandler()) {
            protected void beforeExecute(Thread t, Runnable r) {
                threadPoolBeforeExecute(t, r);
            }
        };
        esScheduled = new ScheduledThreadPoolExecutor(2, new ExecutorThreadFactory(node.threadGroup,
                node.getThreadPoolNamePrefix("scheduled"), classLoader), new RejectionHandler()) {
            protected void beforeExecute(Thread t, Runnable r) {
                threadPoolBeforeExecute(t, r);
            }
        };
        parallelExecutorService = new ParallelExecutorService(node.getLogger(ParallelExecutorService.class.getName()), threadPoolExecutor);
        defaultExecutorService = getOrCreateNamedExecutorService(DEFAULT_EXECUTOR_SERVICE);
        queryExecutorService = getOrCreateNamedExecutorService(QUERY_EXECUTOR_SERVICE, gp.EXECUTOR_QUERY_THREAD_COUNT);
        eventExecutorService = getOrCreateNamedExecutorService(EVENT_EXECUTOR_SERVICE, gp.EXECUTOR_EVENT_THREAD_COUNT);
        mapLoaderExecutorService = parallelExecutorService.newParallelExecutor(gp.MAP_LOAD_THREAD_COUNT.getInteger());
        asyncExecutorService = parallelExecutorService.newBlockingParallelExecutor(24, 1000);
        registerPacketProcessor(EXECUTE, new ExecutionOperationHandler());
        registerPacketProcessor(CANCEL_EXECUTION, new ExecutionCancelOperationHandler());
        started = true;
    }

{'fraction_non_alphanumeric': 0.07505586024781637, 'fraction_numerical': 0.005992281129392647, 'mean_word_length': 4.395616438356164, 'pattern_counts': {'":': 0, '<': 15, '<?xml version=': 0, '>': 15, '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': '1889179', 'n_tokens_mistral': 2616, 'n_tokens_neox': 2560, 'n_words': 639}