<commit_before>/*
 * 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.ignite.internal.processors.cache;

import org.apache.ignite.*;
import org.apache.ignite.cluster.*;
import org.apache.ignite.internal.*;
import org.apache.ignite.internal.cluster.*;
import org.apache.ignite.internal.managers.communication.*;
import org.apache.ignite.internal.managers.deployment.*;
import org.apache.ignite.internal.util.*;
import org.apache.ignite.internal.util.lang.*;
import org.apache.ignite.internal.util.typedef.*;
import org.apache.ignite.internal.util.typedef.internal.*;
import org.apache.ignite.lang.*;
import org.jetbrains.annotations.*;
import org.jsr166.*;

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

import static org.apache.ignite.internal.GridTopic.*;

/**
 * Cache communication manager.
 */
public class GridCacheIoManager extends GridCacheSharedManagerAdapter {
    /** Message ID generator. */
    private static final AtomicLong idGen = new AtomicLong();

    /** Delay in milliseconds between retries. */
    private long retryDelay;

    /** Number of retries using to send messages. */
    private int retryCnt;

    /** Indexed class handlers. */
    private Map<Integer, IgniteBiInClosure[]> idxClsHandlers = new HashMap<>();

    /** Handler registry. */
    private ConcurrentMap<ListenerKey, IgniteBiInClosure<UUID, GridCacheMessage>>
        clsHandlers = new ConcurrentHashMap8<>();

    /** Ordered handler registry. */
    private ConcurrentMap<Object, IgniteBiInClosure<UUID, ? extends GridCacheMessage>> orderedHandlers =
        new ConcurrentHashMap8<>();

    /** Stopping flag. */
    private boolean stopping;

    /** Error flag. */
    private final AtomicBoolean startErr = new AtomicBoolean();

    /** Mutex. */
    private final GridSpinReadWriteLock rw = new GridSpinReadWriteLock();

    /** Deployment enabled. */
    private boolean depEnabled;

    /** Message listener. */
    private GridMessageListener lsnr = new GridMessageListener() {
        @SuppressWarnings("unchecked")
        @Override public void onMessage(final UUID nodeId, Object msg) {
            if (log.isDebugEnabled())
                log.debug("Received unordered cache communication message [nodeId=" + nodeId +
                    ", locId=" + cctx.localNodeId() + ", msg=" + msg + ']');

            final GridCacheMessage cacheMsg = (GridCacheMessage)msg;

            int msgIdx = cacheMsg.lookupIndex();

            IgniteBiInClosure<UUID, GridCacheMessage> c = null;

            if (msgIdx >= 0) {
                IgniteBiInClosure[] cacheClsHandlers = idxClsHandlers.get(cacheMsg.cacheId());

                if (cacheClsHandlers != null)
                    c = cacheClsHandlers[msgIdx];
            }

            if (c == null)
                c = clsHandlers.get(new ListenerKey(cacheMsg.cacheId(), cacheMsg.getClass()));

            if (c == null) {
                if (log.isDebugEnabled())
                    log.debug("Received message without registered handler (will ignore) [msg=" + msg +
                        ", nodeId=" + nodeId + ']');

                return;
            }

            long locTopVer = cctx.discovery().topologyVersion();
            long rmtTopVer = cacheMsg.topologyVersion().topologyVersion();

            if (locTopVer < rmtTopVer) {
                if (log.isDebugEnabled())
                    log.debug("Received message has higher topology version [msg=" + msg +
                        ", locTopVer=" + locTopVer + ", rmtTopVer=" + rmtTopVer + ']');

                IgniteInternalFuture<Long> topFut = cctx.discovery().topologyFuture(rmtTopVer);

                if (!topFut.isDone()) {
                    final IgniteBiInClosure<UUID, GridCacheMessage> c0 = c;

                    topFut.listen(new CI1<IgniteInternalFuture<Long>>() {
                        @Override public void apply(IgniteInternalFuture<Long> t) {
                            onMessage0(nodeId, cacheMsg, c0);
                        }
                    });

                    return;
                }
            }

            onMessage0(nodeId, cacheMsg, c);
        }
    };

    /** {@inheritDoc} */
    @Override public void start0() throws IgniteCheckedException {
        retryDelay = cctx.gridConfig().getNetworkSendRetryDelay();
        retryCnt = cctx.gridConfig().getNetworkSendRetryCount();

        depEnabled = cctx.gridDeploy().enabled();

        cctx.gridIO().addMessageListener(TOPIC_CACHE, lsnr);
    }

    /** {@inheritDoc} */
    @SuppressWarnings("BusyWait")
    @Override protected void onKernalStop0(boolean cancel) {
        cctx.gridIO().removeMessageListener(TOPIC_CACHE);

        for (Object ordTopic : orderedHandlers.keySet())
            cctx.gridIO().removeMessageListener(ordTopic);

        boolean interrupted = false;

        // Busy wait is intentional.
        while (true) {
            try {
                if (rw.tryWriteLock(200, TimeUnit.MILLISECONDS))
                    break;
                else
                    Thread.sleep(200);
            }
            catch (InterruptedException ignore) {
                // Preserve interrupt status & ignore.
                // Note that interrupted flag is cleared.
                interrupted = true;
            }
        }

        if (interrupted)
            Thread.currentThread().interrupt();

        try {
            stopping = true;
        }
        finally {
            rw.writeUnlock();
        }
    }

    /**
     * @param nodeId Node ID.
     * @param cacheMsg Cache message.
     * @param c Handler closure.
     */
    @SuppressWarnings({"unchecked", "ConstantConditions", "ThrowableResultOfMethodCallIgnored"})
    private void onMessage0(final UUID nodeId, final GridCacheMessage cacheMsg,
        final IgniteBiInClosure<UUID, GridCacheMessage> c) {
        rw.readLock();

        try {
            if (stopping) {
                if (log.isDebugEnabled())
                    log.debug("Received cache communication message while stopping (will ignore) [nodeId=" +
                        nodeId + ", msg=" + cacheMsg + ']');

                return;
            }

            if (depEnabled)
                cctx.deploy().ignoreOwnership(true);

            unmarshall(nodeId, cacheMsg);

            if (cacheMsg.allowForStartup())
                processMessage(nodeId, cacheMsg, c);
            else {
                IgniteInternalFuture<?> startFut = startFuture(cacheMsg);

                if (startFut.isDone())
                    processMessage(nodeId, cacheMsg, c);
                else {
                    if (log.isDebugEnabled())
                        log.debug("Waiting for start future to complete for message [nodeId=" + nodeId +
                            ", locId=" + cctx.localNodeId() + ", msg=" + cacheMsg + ']');

                    // Don't hold this thread waiting for preloading to complete.
                    startFut.listen(new CI1<IgniteInternalFuture<?>>() {
                        @Override public void apply(final IgniteInternalFuture<?> f) {
                            cctx.kernalContext().closure().runLocalSafe(
                                new GridPlainRunnable() {
                                    @Override public void run() {
                                        rw.readLock();

                                        try {
                                            if (stopping) {
                                                if (log.isDebugEnabled())
                                                    log.debug("Received cache communication message while stopping " +
                                                        "(will ignore) [nodeId=" + nodeId + ", msg=" + cacheMsg + ']');

                                                return;
                                            }

                                            f.get();

                                            if (log.isDebugEnabled())
                                                log.debug("Start future completed for message [nodeId=" + nodeId +
                                                    ", locId=" + cctx.localNodeId() + ", msg=" + cacheMsg + ']');

                                            processMessage(nodeId, cacheMsg, c);
                                        }
                                        catch (IgniteCheckedException e) {
                                            // Log once.
                                            if (startErr.compareAndSet(false, true))
                                                U.error(log, "Failed to complete preload start future " +
                                                    "(will ignore message) " +
                                                    "[fut=" + f + ", nodeId=" + nodeId + ", msg=" + cacheMsg + ']', e);
                                        }
                                        finally {
                                            rw.readUnlock();
                                        }
                                    }
                                }
                            );
                        }
                    });
                }
            }
        }
        catch (Throwable e) {
            if (X.hasCause(e, ClassNotFoundException.class))
                U.error(log, "Failed to process message (note that distributed services " +
                    "do not support peer class loading, if you deploy distributed service " +
                    "you should have all required classes in CLASSPATH on all nodes in topology) " +
                    "[senderId=" + nodeId + ", err=" + X.cause(e, ClassNotFoundException.class).getMessage() + ']');
            else
                U.error(log, "Failed to process message [senderId=" + nodeId + ']', e);
        }
        finally {
            if (depEnabled)
                cctx.deploy().ignoreOwnership(false);

            rw.readUnlock();
        }
    }

    /**
     * @param cacheMsg Cache message to get start future.
     * @return Preloader start future.
     */
    private IgniteInternalFuture<Object> startFuture(GridCacheMessage cacheMsg) {
        int cacheId = cacheMsg.cacheId();

        return cacheId != 0 ? cctx.cacheContext(cacheId).preloader().startFuture() : cctx.preloadersStartFuture();
    }

    /**
     * @param nodeId Node ID.
     * @param msg Message.
     * @param c Closure.
     */
    private void processMessage(UUID nodeId, GridCacheMessage msg,
        IgniteBiInClosure<UUID, GridCacheMessage> c) {
        try {
            // We will not end up with storing a bunch of new UUIDs
            // in each cache entry, since node ID is stored in NIO session
            // on handshake.
            c.apply(nodeId, msg);

            if (log.isDebugEnabled())
                log.debug("Finished processing cache communication message [nodeId=" + nodeId + ", msg=" + msg + ']');
        }
        catch (Throwable e) {
            U.error(log, "Failed processing message [senderId=" + nodeId + ']', e);
        }
        finally {
            // Reset thread local context.
            cctx.tm().resetContext();
            cctx.mvcc().contextReset();

            // Unwind eviction notifications.
            CU.unwindEvicts(cctx);
        }
    }

    /**
     * Pre-processes message prior to send.
     *
     * @param msg Message to send.
     * @param destNodeId Destination node ID.
     * @throws IgniteCheckedException If failed.
     */
    private void onSend(GridCacheMessage msg, @Nullable UUID destNodeId) throws IgniteCheckedException {
        if (msg.messageId() < 0)<commit_msg># ignite-sprint-3 more info in error message<commit_after>/*
 * 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.ignite.internal.processors.cache;

import org.apache.ignite.*;
import org.apache.ignite.cluster.*;
import org.apache.ignite.internal.*;
import org.apache.ignite.internal.cluster.*;
import org.apache.ignite.internal.managers.communication.*;
import org.apache.ignite.internal.managers.deployment.*;
import org.apache.ignite.internal.util.*;
import org.apache.ignite.internal.util.lang.*;
import org.apache.ignite.internal.util.typedef.*;
import org.apache.ignite.internal.util.typedef.internal.*;
import org.apache.ignite.lang.*;
import org.jetbrains.annotations.*;
import org.jsr166.*;

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

import static org.apache.ignite.internal.GridTopic.*;

/**
 * Cache communication manager.
 */
public class GridCacheIoManager extends GridCacheSharedManagerAdapter {
    /** Message ID generator. */
    private static final AtomicLong idGen = new AtomicLong();

    /** Delay in milliseconds between retries. */
    private long retryDelay;

    /** Number of retries using to send messages. */
    private int retryCnt;

    /** Indexed class handlers. */
    private Map<Integer, IgniteBiInClosure[]> idxClsHandlers = new HashMap<>();

    /** Handler registry. */
    private ConcurrentMap<ListenerKey, IgniteBiInClosure<UUID, GridCacheMessage>>
        clsHandlers = new ConcurrentHashMap8<>();

    /** Ordered handler registry. */
    private ConcurrentMap<Object, IgniteBiInClosure<UUID, ? extends GridCacheMessage>> orderedHandlers =
        new ConcurrentHashMap8<>();

    /** Stopping flag. */
    private boolean stopping;

    /** Error flag. */
    private final AtomicBoolean startErr = new AtomicBoolean();

    /** Mutex. */
    private final GridSpinReadWriteLock rw = new GridSpinReadWriteLock();

    /** Deployment enabled. */
    private boolean depEnabled;

    /** Message listener. */
    private GridMessageListener lsnr = new GridMessageListener() {
        @SuppressWarnings("unchecked")
        @Override public void onMessage(final UUID nodeId, Object msg) {
            if (log.isDebugEnabled())
                log.debug("Received unordered cache communication message [nodeId=" + nodeId +
                    ", locId=" + cctx.localNodeId() + ", msg=" + msg + ']');

            final GridCacheMessage cacheMsg = (GridCacheMessage)msg;

            int msgIdx = cacheMsg.lookupIndex();

            IgniteBiInClosure<UUID, GridCacheMessage> c = null;

            if (msgIdx >= 0) {
                IgniteBiInClosure[] cacheClsHandlers = idxClsHandlers.get(cacheMsg.cacheId());

                if (cacheClsHandlers != null)
                    c = cacheClsHandlers[msgIdx];
            }

            if (c == null)
                c = clsHandlers.get(new ListenerKey(cacheMsg.cacheId(), cacheMsg.getClass()));

            if (c == null) {
                if (log.isDebugEnabled())
                    log.debug("Received message without registered handler (will ignore) [msg=" + msg +
                        ", nodeId=" + nodeId + ']');

                return;
            }

            long locTopVer = cctx.discovery().topologyVersion();
            long rmtTopVer = cacheMsg.topologyVersion().topologyVersion();

            if (locTopVer < rmtTopVer) {
                if (log.isDebugEnabled())
                    log.debug("Received message has higher topology version [msg=" + msg +
                        ", locTopVer=" + locTopVer + ", rmtTopVer=" + rmtTopVer + ']');

                IgniteInternalFuture<Long> topFut = cctx.discovery().topologyFuture(rmtTopVer);

                if (!topFut.isDone()) {
                    final IgniteBiInClosure<UUID, GridCacheMessage> c0 = c;

                    topFut.listen(new CI1<IgniteInternalFuture<Long>>() {
                        @Override public void apply(IgniteInternalFuture<Long> t) {
                            onMessage0(nodeId, cacheMsg, c0);
                        }
                    });

                    return;
                }
            }

            onMessage0(nodeId, cacheMsg, c);
        }
    };

    /** {@inheritDoc} */
    @Override public void start0() throws IgniteCheckedException {
        retryDelay = cctx.gridConfig().getNetworkSendRetryDelay();
        retryCnt = cctx.gridConfig().getNetworkSendRetryCount();

        depEnabled = cctx.gridDeploy().enabled();

        cctx.gridIO().addMessageListener(TOPIC_CACHE, lsnr);
    }

    /** {@inheritDoc} */
    @SuppressWarnings("BusyWait")
    @Override protected void onKernalStop0(boolean cancel) {
        cctx.gridIO().removeMessageListener(TOPIC_CACHE);

        for (Object ordTopic : orderedHandlers.keySet())
            cctx.gridIO().removeMessageListener(ordTopic);

        boolean interrupted = false;

        // Busy wait is intentional.
        while (true) {
            try {
                if (rw.tryWriteLock(200, TimeUnit.MILLISECONDS))
                    break;
                else
                    Thread.sleep(200);
            }
            catch (InterruptedException ignore) {
                // Preserve interrupt status & ignore.
                // Note that interrupted flag is cleared.
                interrupted = true;
            }
        }

        if (interrupted)
            Thread.currentThread().interrupt();

        try {
            stopping = true;
        }
        finally {
            rw.writeUnlock();
        }
    }

    /**
     * @param nodeId Node ID.
     * @param cacheMsg Cache message.
     * @param c Handler closure.
     */
    @SuppressWarnings({"unchecked", "ConstantConditions", "ThrowableResultOfMethodCallIgnored"})
    private void onMessage0(final UUID nodeId, final GridCacheMessage cacheMsg,
        final IgniteBiInClosure<UUID, GridCacheMessage> c) {
        rw.readLock();

        try {
            if (stopping) {
                if (log.isDebugEnabled())
                    log.debug("Received cache communication message while stopping (will ignore) [nodeId=" +
                        nodeId + ", msg=" + cacheMsg + ']');

                return;
            }

            if (depEnabled)
                cctx.deploy().ignoreOwnership(true);

            unmarshall(nodeId, cacheMsg);

            if (cacheMsg.allowForStartup())
                processMessage(nodeId, cacheMsg, c);
            else {
                IgniteInternalFuture<?> startFut = startFuture(cacheMsg);

                if (startFut.isDone())
                    processMessage(nodeId, cacheMsg, c);
                else {
                    if (log.isDebugEnabled())
                        log.debug("Waiting for start future to complete for message [nodeId=" + nodeId +
                            ", locId=" + cctx.localNodeId() + ", msg=" + cacheMsg + ']');

                    // Don't hold this thread waiting for preloading to complete.
                    startFut.listen(new CI1<IgniteInternalFuture<?>>() {
                        @Override public void apply(final IgniteInternalFuture<?> f) {
                            cctx.kernalContext().closure().runLocalSafe(
                                new GridPlainRunnable() {
                                    @Override public void run() {
                                        rw.readLock();

                                        try {
                                            if (stopping) {
                                                if (log.isDebugEnabled())
                                                    log.debug("Received cache communication message while stopping " +
                                                        "(will ignore) [nodeId=" + nodeId + ", msg=" + cacheMsg + ']');

                                                return;
                                            }

                                            f.get();

                                            if (log.isDebugEnabled())
                                                log.debug("Start future completed for message [nodeId=" + nodeId +
                                                    ", locId=" + cctx.localNodeId() + ", msg=" + cacheMsg + ']');

                                            processMessage(nodeId, cacheMsg, c);
                                        }
                                        catch (IgniteCheckedException e) {
                                            // Log once.
                                            if (startErr.compareAndSet(false, true))
                                                U.error(log, "Failed to complete preload start future " +
                                                    "(will ignore message) " +
                                                    "[fut=" + f + ", nodeId=" + nodeId + ", msg=" + cacheMsg + ']', e);
                                        }
                                        finally {
                                            rw.readUnlock();
                                        }
                                    }
                                }
                            );
                        }
                    });
                }
            }
        }
        catch (Throwable e) {
            if (X.hasCause(e, ClassNotFoundException.class))
                U.error(log, "Failed to process message (note that distributed services " +
                    "do not support peer class loading, if you deploy distributed service " +
                    "you should have all required classes in CLASSPATH on all nodes in topology) " +
                    "[senderId=" + nodeId + ", err=" + X.cause(e, ClassNotFoundException.class).getMessage() + ']');
            else
                U.error(log, "Failed to process message [senderId=" + nodeId + ']', e);
        }
        finally {
            if (depEnabled)
                cctx.deploy().ignoreOwnership(false);

            rw.readUnlock();
        }
    }

    /**
     * @param cacheMsg Cache message to get start future.
     * @return Preloader start future.
     */
    private IgniteInternalFuture<Object> startFuture(GridCacheMessage cacheMsg) {
        int cacheId = cacheMsg.cacheId();

        return cacheId != 0 ? cctx.cacheContext(cacheId).preloader().startFuture() : cctx.preloadersStartFuture();
    }

    /**
     * @param nodeId Node ID.
     * @param msg Message.
     * @param c Closure.
     */
    private void processMessage(UUID nodeId, GridCacheMessage msg,
        IgniteBiInClosure<UUID, GridCacheMessage> c) {
        try {
            // We will not end up with storing a bunch of new UUIDs
            // in each cache entry, since node ID is stored in NIO session
            // on handshake.
            c.apply(nodeId, msg);

            if (log.isDebugEnabled())
                log.debug("Finished processing cache communication message [nodeId=" + nodeId + ", msg=" + msg + ']');
        }
        catch (Throwable e) {
            U.error(log, "Failed processing message [senderId=" + nodeId + ", msg=" + msg + ']', e);
        }
        finally {
            // Reset thread local context.
            cctx.tm().resetContext();
            cctx.mvcc().contextReset();

            // Unwind eviction notifications.
            CU.unwindEvicts(cctx);
        }
    }

    /**
     * Pre-processes message prior to send.
     *
     * @param msg Message to send.
     * @param destNodeId Destination node ID.
     * @throws IgniteCheckedException If failed.
     */
    private void onSend(GridCacheMessage msg, @Nullable UUID destNodeId) throws IgniteCheckedException {
        if (msg.messageId() < 0)
{'fraction_non_alphanumeric': 0.10113481978742109, 'fraction_numerical': 0.0021977143770478703, 'mean_word_length': 1.5742645546183913, 'pattern_counts': {'":': 0, '<': 49, '<?xml version=': 0, '>': 47, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 88, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5400950', 'n_tokens_mistral': 6205, 'n_tokens_neox': 5937, 'n_words': 1593}