<commit_before>/*
 * Copyright (c) 2008-2013, Hazelcast, Inc. 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.spi.impl;

import com.hazelcast.core.HazelcastException;
import com.hazelcast.core.HazelcastInstanceNotActiveException;
import com.hazelcast.core.MemberLeftException;
import com.hazelcast.core.PartitionAware;
import com.hazelcast.instance.MemberImpl;
import com.hazelcast.instance.Node;
import com.hazelcast.instance.OutOfMemoryErrorDispatcher;
import com.hazelcast.logging.ILogger;
import com.hazelcast.nio.Address;
import com.hazelcast.nio.Connection;
import com.hazelcast.nio.Packet;
import com.hazelcast.nio.serialization.Data;
import com.hazelcast.partition.PartitionService;
import com.hazelcast.partition.PartitionServiceImpl;
import com.hazelcast.partition.InternalPartition;
import com.hazelcast.spi.*;
import com.hazelcast.spi.annotation.PrivateApi;
import com.hazelcast.spi.exception.CallTimeoutException;
import com.hazelcast.spi.exception.CallerNotMemberException;
import com.hazelcast.spi.exception.PartitionMigratingException;
import com.hazelcast.spi.exception.WrongTargetException;
import com.hazelcast.spi.impl.PartitionIteratingOperation.PartitionResponse;
import com.hazelcast.util.Clock;
import com.hazelcast.util.executor.AbstractExecutorThreadFactory;
import com.hazelcast.util.executor.ManagedExecutorService;
import com.hazelcast.util.executor.SingleExecutorThreadFactory;
import com.hazelcast.util.scheduler.*;

import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicLong;

/**
 * This is the Basic InternalOperationService and depends on Java 6.
 *
 * All the classes that begin with 'Basic' are implementation detail that depend on the
 * {@link com.hazelcast.spi.impl.BasicOperationService}.
 *
 * <h1>System Operation</h1>
 * When a {@link com.hazelcast.spi.UrgentSystemOperation} is invoked on this OperationService, it will be executed with a
 * high urgency by making use of a urgent queue. So when the system is under load, and the operation queues are
 * filled, then system operations are executed before normal operation. The advantage is that when a system is under
 * pressure, it still is able to do things like recognizing new members in the cluster and moving partitions around.
 *
 * When a UrgentSystemOperation is send to a remote machine, it is wrapped in a {@link Packet} and the packet is marked as a
 * urgent packet. When this packet is received on the remove OperationService, the urgent flag is checked and if
 * needed, the operation is set on the urgent queue. So local and remote execution of System operations will obey
 * the urgency.
 *
 * @author mdogan 12/14/12
 * @see com.hazelcast.spi.impl.BasicInvocation
 * @see com.hazelcast.spi.impl.BasicInvocationBuilder
 * @see com.hazelcast.spi.impl.BasicPartitionInvocation
 * @see com.hazelcast.spi.impl.BasicTargetInvocation
 */
final class BasicOperationService implements InternalOperationService {

    private final AtomicLong executedOperationsCount = new AtomicLong();

    private final NodeEngineImpl nodeEngine;
    private final Node node;
    private final ILogger logger;
    private final AtomicLong callIdGen = new AtomicLong(0);
    private final ConcurrentMap<Long, RemoteCall> remoteCalls;
    private final ExecutorService[] operationExecutors;
    private final BlockingQueue[] operationExecutorQueues;

    private final ExecutorService defaultOperationExecutor;
    private final ConcurrentLinkedQueue defaultOperationUrgentQueue;
    private final ExecutorService responseExecutor;
    private final long defaultCallTimeout;
    private final Map<RemoteCallKey, RemoteCallKey> executingCalls;
    private final ConcurrentMap<Long, BackupCompletionCallback> backupCalls;
    private final int operationThreadCount;
    private final EntryTaskScheduler<Object, ScheduledBackup> backupScheduler;
    private final BlockingQueue<Runnable> responseWorkQueue = new LinkedBlockingQueue<Runnable>();
    private final ConcurrentLinkedQueue[] operationExecutorUrgentQueues;
    private final ExecutionService executionService;

    BasicOperationService(NodeEngineImpl nodeEngine) {
        this.nodeEngine = nodeEngine;
        this.node = nodeEngine.getNode();
        this.logger = node.getLogger(OperationService.class.getName());
        defaultCallTimeout = node.getGroupProperties().OPERATION_CALL_TIMEOUT_MILLIS.getLong();
        final int coreSize = Runtime.getRuntime().availableProcessors();
        final boolean reallyMultiCore = coreSize >= 8;
        final int concurrencyLevel = reallyMultiCore ? coreSize * 4 : 16;
        remoteCalls = new ConcurrentHashMap<Long, RemoteCall>(1000, 0.75f, concurrencyLevel);
        final int opThreadCount = node.getGroupProperties().OPERATION_THREAD_COUNT.getInteger();
        operationThreadCount =  opThreadCount > 0 ? opThreadCount : coreSize * 2;
        operationExecutors = new ExecutorService[operationThreadCount];
        operationExecutorQueues = new BlockingQueue[operationThreadCount];
        operationExecutorUrgentQueues = new ConcurrentLinkedQueue[operationThreadCount];
        for (int i = 0; i < operationExecutors.length; i++) {
            BlockingQueue<Runnable> operationExecutorQueue = new LinkedBlockingQueue<Runnable>();
            operationExecutorQueues[i] = operationExecutorQueue;

            ConcurrentLinkedQueue<Runnable> operationExecutorUrgentQueue = new ConcurrentLinkedQueue<Runnable>();
            operationExecutorUrgentQueues[i]=operationExecutorUrgentQueue;

            operationExecutors[i] = new ThreadPoolExecutor(1, 1,
                    0L, TimeUnit.MILLISECONDS,
                    operationExecutorQueue,
                    new OperationThreadFactory(i));
        }

        executionService = nodeEngine.getExecutionService();
        defaultOperationUrgentQueue = new ConcurrentLinkedQueue<Runnable>();
        defaultOperationExecutor = executionService.register(ExecutionService.OPERATION_EXECUTOR,
                coreSize * 2, coreSize * 100000);

        executionService.register(ExecutionService.ASYNC_EXECUTOR, coreSize * 10, coreSize * 100000);

        responseExecutor = new ThreadPoolExecutor(1, 1,
                0L, TimeUnit.MILLISECONDS,
                responseWorkQueue,
                new SingleExecutorThreadFactory(node.threadGroup,
                        node.getConfigClassLoader(), node.getThreadNamePrefix("response")));

        executingCalls = new ConcurrentHashMap<RemoteCallKey, RemoteCallKey>(1000, 0.75f, concurrencyLevel);
        backupCalls = new ConcurrentHashMap<Long, BackupCompletionCallback>(1000, 0.75f, concurrencyLevel);
        backupScheduler = EntryTaskSchedulerFactory.newScheduler(executionService.getScheduledExecutor(),
                new ScheduledBackupProcessor(), ScheduleType.SCHEDULE_IF_NEW);
    }

    @Override
    public int getOperationThreadCount(){
        return operationThreadCount;
    }

    @Override
    public int getRunningOperationsCount(){
        return executingCalls.size();
    }

    @Override
    public long getExecutedOperationCount(){
        return executedOperationsCount.get();
    }

    @Override
    public int getRemoteOperationsCount(){
        return remoteCalls.size();
    }

    @Override
    public int getResponseQueueSize() {
        return responseWorkQueue.size();
    }

    @Override
    public int getOperationExecutorQueueSize() {
        int size = 0;
        for(BlockingQueue q: operationExecutorQueues){
            size+=q.size();
        }
        return size;
    }

    @Override
    public InvocationBuilder createInvocationBuilder(String serviceName, Operation op, final int partitionId) {
        if (partitionId < 0) throw new IllegalArgumentException("Partition id cannot be negative!");
        return new BasicInvocationBuilder(nodeEngine, serviceName, op, partitionId);
    }

    @Override
    public InvocationBuilder createInvocationBuilder(String serviceName, Operation op, Address target) {
        if (target == null) throw new IllegalArgumentException("Target cannot be null!");
        return new BasicInvocationBuilder(nodeEngine, serviceName, op, target);
    }

    @PrivateApi
    public void handleOperation(final Packet packet) {
        try {
            if (packet.isHeaderSet(Packet.HEADER_RESPONSE)) {
                responseExecutor.execute(new ResponseProcessor(packet));
                //responseExecutor.execute(new RemoteOperationProcessor(packet));
            } else {
                final int partitionId = packet.getPartitionId();
                final Executor executor = getExecutor(partitionId);
                if(packet.isUrgent()){
                    ConcurrentLinkedQueue<Runnable> urgentQueue = getUrgentQueue(partitionId);
                    urgentQueue.add(new RemoteOperationProcessor(packet));
                    executor.execute(new UrgentSystemOperationsProcessor());
                } else{
                    executor.execute(new RemoteOperationProcessor(packet));
                }
            }
        } catch (RejectedExecutionException e) {
            if (nodeEngine.isActive()) {
                throw e;
            }
        }<commit_msg>Removed piece of commented out code<commit_after>/*
 * Copyright (c) 2008-2013, Hazelcast, Inc. 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.spi.impl;

import com.hazelcast.core.HazelcastException;
import com.hazelcast.core.HazelcastInstanceNotActiveException;
import com.hazelcast.core.MemberLeftException;
import com.hazelcast.core.PartitionAware;
import com.hazelcast.instance.MemberImpl;
import com.hazelcast.instance.Node;
import com.hazelcast.instance.OutOfMemoryErrorDispatcher;
import com.hazelcast.logging.ILogger;
import com.hazelcast.nio.Address;
import com.hazelcast.nio.Connection;
import com.hazelcast.nio.Packet;
import com.hazelcast.nio.serialization.Data;
import com.hazelcast.partition.PartitionService;
import com.hazelcast.partition.PartitionServiceImpl;
import com.hazelcast.partition.InternalPartition;
import com.hazelcast.spi.*;
import com.hazelcast.spi.annotation.PrivateApi;
import com.hazelcast.spi.exception.CallTimeoutException;
import com.hazelcast.spi.exception.CallerNotMemberException;
import com.hazelcast.spi.exception.PartitionMigratingException;
import com.hazelcast.spi.exception.WrongTargetException;
import com.hazelcast.spi.impl.PartitionIteratingOperation.PartitionResponse;
import com.hazelcast.util.Clock;
import com.hazelcast.util.executor.AbstractExecutorThreadFactory;
import com.hazelcast.util.executor.ManagedExecutorService;
import com.hazelcast.util.executor.SingleExecutorThreadFactory;
import com.hazelcast.util.scheduler.*;

import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicLong;

/**
 * This is the Basic InternalOperationService and depends on Java 6.
 *
 * All the classes that begin with 'Basic' are implementation detail that depend on the
 * {@link com.hazelcast.spi.impl.BasicOperationService}.
 *
 * <h1>System Operation</h1>
 * When a {@link com.hazelcast.spi.UrgentSystemOperation} is invoked on this OperationService, it will be executed with a
 * high urgency by making use of a urgent queue. So when the system is under load, and the operation queues are
 * filled, then system operations are executed before normal operation. The advantage is that when a system is under
 * pressure, it still is able to do things like recognizing new members in the cluster and moving partitions around.
 *
 * When a UrgentSystemOperation is send to a remote machine, it is wrapped in a {@link Packet} and the packet is marked as a
 * urgent packet. When this packet is received on the remove OperationService, the urgent flag is checked and if
 * needed, the operation is set on the urgent queue. So local and remote execution of System operations will obey
 * the urgency.
 *
 * @author mdogan 12/14/12
 * @see com.hazelcast.spi.impl.BasicInvocation
 * @see com.hazelcast.spi.impl.BasicInvocationBuilder
 * @see com.hazelcast.spi.impl.BasicPartitionInvocation
 * @see com.hazelcast.spi.impl.BasicTargetInvocation
 */
final class BasicOperationService implements InternalOperationService {

    private final AtomicLong executedOperationsCount = new AtomicLong();

    private final NodeEngineImpl nodeEngine;
    private final Node node;
    private final ILogger logger;
    private final AtomicLong callIdGen = new AtomicLong(0);
    private final ConcurrentMap<Long, RemoteCall> remoteCalls;
    private final ExecutorService[] operationExecutors;
    private final BlockingQueue[] operationExecutorQueues;

    private final ExecutorService defaultOperationExecutor;
    private final ConcurrentLinkedQueue defaultOperationUrgentQueue;
    private final ExecutorService responseExecutor;
    private final long defaultCallTimeout;
    private final Map<RemoteCallKey, RemoteCallKey> executingCalls;
    private final ConcurrentMap<Long, BackupCompletionCallback> backupCalls;
    private final int operationThreadCount;
    private final EntryTaskScheduler<Object, ScheduledBackup> backupScheduler;
    private final BlockingQueue<Runnable> responseWorkQueue = new LinkedBlockingQueue<Runnable>();
    private final ConcurrentLinkedQueue[] operationExecutorUrgentQueues;
    private final ExecutionService executionService;

    BasicOperationService(NodeEngineImpl nodeEngine) {
        this.nodeEngine = nodeEngine;
        this.node = nodeEngine.getNode();
        this.logger = node.getLogger(OperationService.class.getName());
        defaultCallTimeout = node.getGroupProperties().OPERATION_CALL_TIMEOUT_MILLIS.getLong();
        final int coreSize = Runtime.getRuntime().availableProcessors();
        final boolean reallyMultiCore = coreSize >= 8;
        final int concurrencyLevel = reallyMultiCore ? coreSize * 4 : 16;
        remoteCalls = new ConcurrentHashMap<Long, RemoteCall>(1000, 0.75f, concurrencyLevel);
        final int opThreadCount = node.getGroupProperties().OPERATION_THREAD_COUNT.getInteger();
        operationThreadCount =  opThreadCount > 0 ? opThreadCount : coreSize * 2;
        operationExecutors = new ExecutorService[operationThreadCount];
        operationExecutorQueues = new BlockingQueue[operationThreadCount];
        operationExecutorUrgentQueues = new ConcurrentLinkedQueue[operationThreadCount];
        for (int i = 0; i < operationExecutors.length; i++) {
            BlockingQueue<Runnable> operationExecutorQueue = new LinkedBlockingQueue<Runnable>();
            operationExecutorQueues[i] = operationExecutorQueue;

            ConcurrentLinkedQueue<Runnable> operationExecutorUrgentQueue = new ConcurrentLinkedQueue<Runnable>();
            operationExecutorUrgentQueues[i]=operationExecutorUrgentQueue;

            operationExecutors[i] = new ThreadPoolExecutor(1, 1,
                    0L, TimeUnit.MILLISECONDS,
                    operationExecutorQueue,
                    new OperationThreadFactory(i));
        }

        executionService = nodeEngine.getExecutionService();
        defaultOperationUrgentQueue = new ConcurrentLinkedQueue<Runnable>();
        defaultOperationExecutor = executionService.register(ExecutionService.OPERATION_EXECUTOR,
                coreSize * 2, coreSize * 100000);

        executionService.register(ExecutionService.ASYNC_EXECUTOR, coreSize * 10, coreSize * 100000);

        responseExecutor = new ThreadPoolExecutor(1, 1,
                0L, TimeUnit.MILLISECONDS,
                responseWorkQueue,
                new SingleExecutorThreadFactory(node.threadGroup,
                        node.getConfigClassLoader(), node.getThreadNamePrefix("response")));

        executingCalls = new ConcurrentHashMap<RemoteCallKey, RemoteCallKey>(1000, 0.75f, concurrencyLevel);
        backupCalls = new ConcurrentHashMap<Long, BackupCompletionCallback>(1000, 0.75f, concurrencyLevel);
        backupScheduler = EntryTaskSchedulerFactory.newScheduler(executionService.getScheduledExecutor(),
                new ScheduledBackupProcessor(), ScheduleType.SCHEDULE_IF_NEW);
    }

    @Override
    public int getOperationThreadCount(){
        return operationThreadCount;
    }

    @Override
    public int getRunningOperationsCount(){
        return executingCalls.size();
    }

    @Override
    public long getExecutedOperationCount(){
        return executedOperationsCount.get();
    }

    @Override
    public int getRemoteOperationsCount(){
        return remoteCalls.size();
    }

    @Override
    public int getResponseQueueSize() {
        return responseWorkQueue.size();
    }

    @Override
    public int getOperationExecutorQueueSize() {
        int size = 0;
        for(BlockingQueue q: operationExecutorQueues){
            size+=q.size();
        }
        return size;
    }

    @Override
    public InvocationBuilder createInvocationBuilder(String serviceName, Operation op, final int partitionId) {
        if (partitionId < 0) throw new IllegalArgumentException("Partition id cannot be negative!");
        return new BasicInvocationBuilder(nodeEngine, serviceName, op, partitionId);
    }

    @Override
    public InvocationBuilder createInvocationBuilder(String serviceName, Operation op, Address target) {
        if (target == null) throw new IllegalArgumentException("Target cannot be null!");
        return new BasicInvocationBuilder(nodeEngine, serviceName, op, target);
    }

    @PrivateApi
    public void handleOperation(final Packet packet) {
        try {
            if (packet.isHeaderSet(Packet.HEADER_RESPONSE)) {
                responseExecutor.execute(new ResponseProcessor(packet));
            } else {
                final int partitionId = packet.getPartitionId();
                final Executor executor = getExecutor(partitionId);
                if(packet.isUrgent()){
                    ConcurrentLinkedQueue<Runnable> urgentQueue = getUrgentQueue(partitionId);
                    urgentQueue.add(new RemoteOperationProcessor(packet));
                    executor.execute(new UrgentSystemOperationsProcessor());
                } else{
                    executor.execute(new RemoteOperationProcessor(packet));
                }
            }
        } catch (RejectedExecutionException e) {
            if (nodeEngine.isActive()) {
                throw e;
            }
        }
{'fraction_non_alphanumeric': 0.07570458753342933, 'fraction_numerical': 0.00750874305698416, 'mean_word_length': 3.7507940386024923, 'pattern_counts': {'":': 0, '<': 41, '<?xml version=': 0, '>': 41, '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': '4906750', 'n_tokens_mistral': 5135, 'n_tokens_neox': 4914, 'n_words': 1345}