<commit_before>/**
 * Copyright (c) 2016-2021 by the respective copyright holders.
 * All rights reserved. This program and the accompanying materials
 * are made available under the terms of the Eclipse Public License v1.0
 * which accompanies this distribution, and is available at
 * http://www.eclipse.org/legal/epl-v10.html
 */
package com.zsmartsystems.zigbee.transaction;

import java.util.ArrayList;
import java.util.Collections;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Future;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.zsmartsystems.zigbee.CommandResult;
import com.zsmartsystems.zigbee.IeeeAddress;
import com.zsmartsystems.zigbee.ZigBeeAddress;
import com.zsmartsystems.zigbee.ZigBeeBroadcastDestination;
import com.zsmartsystems.zigbee.ZigBeeCommand;
import com.zsmartsystems.zigbee.ZigBeeEndpointAddress;
import com.zsmartsystems.zigbee.ZigBeeExecutors;
import com.zsmartsystems.zigbee.ZigBeeNetworkManager;
import com.zsmartsystems.zigbee.ZigBeeNetworkNodeListener;
import com.zsmartsystems.zigbee.ZigBeeNode;
import com.zsmartsystems.zigbee.transaction.ZigBeeTransaction.TransactionState;
import com.zsmartsystems.zigbee.transport.ZigBeeTransportProgressState;
import com.zsmartsystems.zigbee.zdo.field.NodeDescriptor.MacCapabilitiesType;

/**
 * The centralised transaction manager is designed to solve a number of issues associated with the sending
 * of transactions to devices, and ensure a smooth flow of data under most circumstances. It sits between the
 * applications that are sending commands with no knowledge of other applications, or knowledge of the network state,
 * and the transport layer.
 * <p>
 * The following issues are covered -:
 * <ul>
 * <li>ZigBee specifies a maximum rate at which broadcast transactions can be sent. This is primarily designed to limit
 * the amount of memory required in a ZigBee router to buffer frames to sleepy child devices.
 * <li>Sending multiple transactions to a single device without waiting for the response to the first transactions may
 * result in lost transactions.
 * <li>Retry management. Each queue can be configured to allow retries. The transaction futures are not completed until
 * the transaction completes successfully, or the maximum number of retries are attempted.
 * <li>Account for transport layer feedback.
 * <li>Manage sleepy devices. The queue manager will only send a certain number of transactions to sleepy devices at
 * once. This ensures that there is always bandwidth reserved for mains powered devices, thus ensuring that the queue
 * can not be filled with commands to sleepy devices, which would adversely affect the overall performance of the
 * system.
 * </ul>
 * <p>
 * Each time a transaction is sent to the transaction manager, or a transaction completes, the transaction manager will
 * attempt to send the next transaction in order to keep the transport layer full, while also fulfilling the various
 * constraints in the queues.
 * <p>
 * When sending, queues are polled in random order to ensure that all queues get a fair chance at sending data.
 *
 * @author <NAME>
 *
 */
public class ZigBeeTransactionManager implements ZigBeeNetworkNodeListener {
    /**
     * The logger.
     */
    private final Logger logger = LoggerFactory.getLogger(ZigBeeTransactionManager.class);

    /**
     * The maximum number of outstanding transactions from all queues
     */
    private final int MAX_OUTSTANDING_TRANSACTIONS = 5;

    /**
     * The maximum number of outstanding transactions from sleepy queues
     */
    private final int MAX_SLEEPY_TRANSACTIONS = 3;

    private final int NODE_RETRIES = 2;
    private final int NODE_TRANSACTIONS = 1;
    private final int NODE_DELAY = 50;

    private final int SLEEPY_RETRIES = 2;
    private final int SLEEPY_TRANSACTIONS = 1;
    private final int SLEEPY_DELAY = 50;

    private final int MCAST_RETRIES = 0;
    private final int MCAST_TRANSACTIONS = 3;
    private final int MCAST_DELAY = 1200;

    private final int BCAST_RETRIES = 0;
    private final int BCAST_TRANSACTIONS = 2;
    private final int BCAST_DELAY = 4000;

    /**
     * The {@link ZigBeeNetworkManager} to which this manager belongs
     */
    private final ZigBeeNetworkManager networkManager;

    /**
     * The set of outstanding transactions - used to notify transactions when responses are received.
     */
    private final Set<ZigBeeTransaction> outstandingTransactions = new HashSet<>();

    /**
     * The maximum number of transactions the manager will allow at any time
     */
    private int maxOutstandingTransactions = MAX_OUTSTANDING_TRANSACTIONS;

    /**
     * The maximum number of outstanding transactions from sleepy queues
     */
    private int maxSleepyTransactions = MAX_SLEEPY_TRANSACTIONS;

    /**
     * A counter holding the number of sleepy transactions
     */
    private int sleepyTransactions = 0;

    /**
     * Flag set to true after the shutdown method has been called
     */
    private boolean isShutdown = false;

    /**
     * Executor service to execute update threads for discovery or mesh updates etc.
     * We use a {@link ZigBeeExecutors.newScheduledThreadPool} to provide a fixed number of threads as otherwise this
     * could result in a large number of simultaneous threads in large networks. The threads are only used to time out a
     * transaction which is a short running event so should not block other threads from running in any practical sense.
     */
    private final ScheduledExecutorService executorService = ZigBeeExecutors.newScheduledThreadPool(6,
            "TransactionManager");

    /**
     * A Map containing the queue for each node. This provides quick access when adding commands to queue, or performing
     * any queue function where we know the node.
     */
    private final Map<IeeeAddress, ZigBeeTransactionQueue> nodeQueue = new ConcurrentHashMap<>();

    /**
     * {@link AtomicInteger} used to generate transaction IDs
     */
    private final AtomicInteger transactionIdCounter = new AtomicInteger();

    /**
     * A List of queues with outstanding commands. This is used for random access when sending transactions.
     */
    private final List<ZigBeeTransactionQueue> outstandingQueues = new ArrayList<>();

    private final ZigBeeTransactionQueue defaultQueue;
    private final ZigBeeTransactionQueue broadcastQueue;
    private final ZigBeeTransactionQueue multicastQueue;

    private ZigBeeTransactionProfile defaultProfile;
    private ZigBeeTransactionProfile defaultSleepyProfile;

    /**
     * Timer used to keep the queue running
     */
    private ScheduledFuture<?> timeoutTask;

    public ZigBeeTransactionManager(ZigBeeNetworkManager manager) {
        this.networkManager = manager;

        defaultProfile = new ZigBeeTransactionProfile(NODE_RETRIES, NODE_TRANSACTIONS, NODE_DELAY);
        defaultSleepyProfile = new ZigBeeTransactionProfile(SLEEPY_RETRIES, SLEEPY_TRANSACTIONS, SLEEPY_DELAY);

        defaultQueue = new ZigBeeTransactionQueue("Default");
        defaultQueue.setProfile(defaultProfile);
        defaultQueue.setSleepy(false);

        broadcastQueue = new ZigBeeTransactionQueue("Broadcast");
        broadcastQueue.setProfile(new ZigBeeTransactionProfile(BCAST_RETRIES, BCAST_TRANSACTIONS, BCAST_DELAY));

        multicastQueue = new ZigBeeTransactionQueue("Multicast");
        multicastQueue.setProfile(new ZigBeeTransactionProfile(MCAST_RETRIES, MCAST_TRANSACTIONS, MCAST_DELAY));<commit_msg>Use thread-safe list type for outstandingQueues (#1291)<commit_after>/**
 * Copyright (c) 2016-2021 by the respective copyright holders.
 * All rights reserved. This program and the accompanying materials
 * are made available under the terms of the Eclipse Public License v1.0
 * which accompanies this distribution, and is available at
 * http://www.eclipse.org/legal/epl-v10.html
 */
package com.zsmartsystems.zigbee.transaction;

import java.util.ArrayList;
import java.util.Collections;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.Future;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.zsmartsystems.zigbee.CommandResult;
import com.zsmartsystems.zigbee.IeeeAddress;
import com.zsmartsystems.zigbee.ZigBeeAddress;
import com.zsmartsystems.zigbee.ZigBeeBroadcastDestination;
import com.zsmartsystems.zigbee.ZigBeeCommand;
import com.zsmartsystems.zigbee.ZigBeeEndpointAddress;
import com.zsmartsystems.zigbee.ZigBeeExecutors;
import com.zsmartsystems.zigbee.ZigBeeNetworkManager;
import com.zsmartsystems.zigbee.ZigBeeNetworkNodeListener;
import com.zsmartsystems.zigbee.ZigBeeNode;
import com.zsmartsystems.zigbee.transaction.ZigBeeTransaction.TransactionState;
import com.zsmartsystems.zigbee.transport.ZigBeeTransportProgressState;
import com.zsmartsystems.zigbee.zdo.field.NodeDescriptor.MacCapabilitiesType;

/**
 * The centralised transaction manager is designed to solve a number of issues associated with the sending
 * of transactions to devices, and ensure a smooth flow of data under most circumstances. It sits between the
 * applications that are sending commands with no knowledge of other applications, or knowledge of the network state,
 * and the transport layer.
 * <p>
 * The following issues are covered -:
 * <ul>
 * <li>ZigBee specifies a maximum rate at which broadcast transactions can be sent. This is primarily designed to limit
 * the amount of memory required in a ZigBee router to buffer frames to sleepy child devices.
 * <li>Sending multiple transactions to a single device without waiting for the response to the first transactions may
 * result in lost transactions.
 * <li>Retry management. Each queue can be configured to allow retries. The transaction futures are not completed until
 * the transaction completes successfully, or the maximum number of retries are attempted.
 * <li>Account for transport layer feedback.
 * <li>Manage sleepy devices. The queue manager will only send a certain number of transactions to sleepy devices at
 * once. This ensures that there is always bandwidth reserved for mains powered devices, thus ensuring that the queue
 * can not be filled with commands to sleepy devices, which would adversely affect the overall performance of the
 * system.
 * </ul>
 * <p>
 * Each time a transaction is sent to the transaction manager, or a transaction completes, the transaction manager will
 * attempt to send the next transaction in order to keep the transport layer full, while also fulfilling the various
 * constraints in the queues.
 * <p>
 * When sending, queues are polled in random order to ensure that all queues get a fair chance at sending data.
 *
 * @author <NAME>
 *
 */
public class ZigBeeTransactionManager implements ZigBeeNetworkNodeListener {
    /**
     * The logger.
     */
    private final Logger logger = LoggerFactory.getLogger(ZigBeeTransactionManager.class);

    /**
     * The maximum number of outstanding transactions from all queues
     */
    private final int MAX_OUTSTANDING_TRANSACTIONS = 5;

    /**
     * The maximum number of outstanding transactions from sleepy queues
     */
    private final int MAX_SLEEPY_TRANSACTIONS = 3;

    private final int NODE_RETRIES = 2;
    private final int NODE_TRANSACTIONS = 1;
    private final int NODE_DELAY = 50;

    private final int SLEEPY_RETRIES = 2;
    private final int SLEEPY_TRANSACTIONS = 1;
    private final int SLEEPY_DELAY = 50;

    private final int MCAST_RETRIES = 0;
    private final int MCAST_TRANSACTIONS = 3;
    private final int MCAST_DELAY = 1200;

    private final int BCAST_RETRIES = 0;
    private final int BCAST_TRANSACTIONS = 2;
    private final int BCAST_DELAY = 4000;

    /**
     * The {@link ZigBeeNetworkManager} to which this manager belongs
     */
    private final ZigBeeNetworkManager networkManager;

    /**
     * The set of outstanding transactions - used to notify transactions when responses are received.
     */
    private final Set<ZigBeeTransaction> outstandingTransactions = new HashSet<>();

    /**
     * The maximum number of transactions the manager will allow at any time
     */
    private int maxOutstandingTransactions = MAX_OUTSTANDING_TRANSACTIONS;

    /**
     * The maximum number of outstanding transactions from sleepy queues
     */
    private int maxSleepyTransactions = MAX_SLEEPY_TRANSACTIONS;

    /**
     * A counter holding the number of sleepy transactions
     */
    private int sleepyTransactions = 0;

    /**
     * Flag set to true after the shutdown method has been called
     */
    private boolean isShutdown = false;

    /**
     * Executor service to execute update threads for discovery or mesh updates etc.
     * We use a {@link ZigBeeExecutors.newScheduledThreadPool} to provide a fixed number of threads as otherwise this
     * could result in a large number of simultaneous threads in large networks. The threads are only used to time out a
     * transaction which is a short running event so should not block other threads from running in any practical sense.
     */
    private final ScheduledExecutorService executorService = ZigBeeExecutors.newScheduledThreadPool(6,
            "TransactionManager");

    /**
     * A Map containing the queue for each node. This provides quick access when adding commands to queue, or performing
     * any queue function where we know the node.
     */
    private final Map<IeeeAddress, ZigBeeTransactionQueue> nodeQueue = new ConcurrentHashMap<>();

    /**
     * {@link AtomicInteger} used to generate transaction IDs
     */
    private final AtomicInteger transactionIdCounter = new AtomicInteger();

    /**
     * A List of queues with outstanding commands. This is used for random access when sending transactions.
     */
    private final List<ZigBeeTransactionQueue> outstandingQueues = new CopyOnWriteArrayList<>();

    private final ZigBeeTransactionQueue defaultQueue;
    private final ZigBeeTransactionQueue broadcastQueue;
    private final ZigBeeTransactionQueue multicastQueue;

    private ZigBeeTransactionProfile defaultProfile;
    private ZigBeeTransactionProfile defaultSleepyProfile;

    /**
     * Timer used to keep the queue running
     */
    private ScheduledFuture<?> timeoutTask;

    public ZigBeeTransactionManager(ZigBeeNetworkManager manager) {
        this.networkManager = manager;

        defaultProfile = new ZigBeeTransactionProfile(NODE_RETRIES, NODE_TRANSACTIONS, NODE_DELAY);
        defaultSleepyProfile = new ZigBeeTransactionProfile(SLEEPY_RETRIES, SLEEPY_TRANSACTIONS, SLEEPY_DELAY);

        defaultQueue = new ZigBeeTransactionQueue("Default");
        defaultQueue.setProfile(defaultProfile);
        defaultQueue.setSleepy(false);

        broadcastQueue = new ZigBeeTransactionQueue("Broadcast");
        broadcastQueue.setProfile(new ZigBeeTransactionProfile(BCAST_RETRIES, BCAST_TRANSACTIONS, BCAST_DELAY));

        multicastQueue = new ZigBeeTransactionQueue("Multicast");
        multicastQueue.setProfile(new ZigBeeTransactionProfile(MCAST_RETRIES, MCAST_TRANSACTIONS, MCAST_DELAY));
{'fraction_non_alphanumeric': 0.0619463466733681, 'fraction_numerical': 0.005026072752403091, 'mean_word_length': 4.13649564375605, 'pattern_counts': {'":': 0, '<': 39, '<?xml version=': 0, '>': 39, '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': '3442444', 'n_tokens_mistral': 4353, 'n_tokens_neox': 4124, 'n_words': 1531}