<commit_before>/*
 * Tigase Jabber/XMPP Server
 * Copyright (C) 2004-2008 "<NAME>" <<EMAIL>>
 *
 * This program is free software: you can redistribute it and/or modify
 * it under the terms of the GNU General Public License as published by
 * the Free Software Foundation, version 3 of the License.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with this program. Look for COPYING file in the top folder.
 * If not, see http://www.gnu.org/licenses/.
 *
 * $Rev$
 * Last modified by $Author$
 * $Date$
 */

package tigase.util;

//~--- non-JDK imports --------------------------------------------------------

import tigase.server.Packet;

import tigase.xmpp.XMPPProcessorIfc;
import tigase.xmpp.XMPPResourceConnection;

//~--- JDK imports ------------------------------------------------------------

import java.util.ArrayList;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.logging.Level;
import java.util.logging.Logger;

//~--- classes ----------------------------------------------------------------

/**
 * Created: Apr 21, 2009 8:50:50 PM
 *
 * @param <E>
 * @author <a href="mailto:<EMAIL>"><NAME></a>
 * @version $Rev$
 */
public class ProcessingThreads<E extends WorkerThread> {
	private static final Logger log = Logger.getLogger(ProcessingThreads.class.getName());

	//~--- fields ---------------------------------------------------------------

	private long droppedPackets = 0;

	// private int maxQueueSize = 10000;
	private String name = null;

	// Packets are put in queues in such a way that all packets for the same
	// user end-up in the same queue. This is important in some cases as
	// packet processing order does matter in some cases, especially for
	// roster processing.
	// Therefore it is also recommended that there is a single thread for
	// each queue but we can ditribute load increasing number of queues.
	// private int numQueues = 2;
	private int numWorkerThreads = 1;
	private ArrayList<E> workerThreads = null;

	//~--- constructors ---------------------------------------------------------

//private ArrayList<PriorityQueueAbstract<QueueItem>> queues =
//  new ArrayList<PriorityQueueAbstract<QueueItem>>(1);

	/**
	 * Constructs ...
	 *
	 *
	 * @param worker
	 * @param numWorkerThreads
	 * @param maxQueueSize
	 * @param name
	 * @throws ClassNotFoundException
	 * @throws InstantiationException
	 * @throws IllegalAccessException
	 */
	@SuppressWarnings({ "unchecked" })
	public ProcessingThreads(E worker, int numWorkerThreads, int maxQueueSize, String name)
			throws ClassNotFoundException, InstantiationException, IllegalAccessException {

		// this.numQueues = numQueues;
		// this.maxQueueSize = maxQueueSize;
		this.numWorkerThreads = numWorkerThreads;
		workerThreads = new ArrayList<E>(numWorkerThreads);
		this.name = name;

//  for (int i = 0; i < numQueues; i++) {
		// LinkedBlockingQueue<QueueItem> queue = new LinkedBlockingQueue<QueueItem>(maxQueueSize);
//  queues.add(queue);
		for (int j = 0; j < numWorkerThreads; j++) {
			WorkerThread t = worker.getNewInstance();

			t.setQueueMaxSize(maxQueueSize);
			t.setDaemon(true);
			t.setName(name + " Queue Worker " + j);
			t.start();
			workerThreads.add((E) t);
		}

//  }
	}

	//~--- methods --------------------------------------------------------------

	/**
	 * Method description
	 *
	 *
	 * @param processor
	 * @param packet
	 * @param conn
	 *
	 * @return
	 */
	public boolean addItem(XMPPProcessorIfc processor, Packet packet, XMPPResourceConnection conn) {
		boolean ret = false;
		QueueItem item = new QueueItem(processor, packet, conn);

		try {
			if ((item.getConn() != null) && item.getConn().isAuthorized()) {

				// Queueing packets per user...
				ret = workerThreads.get(Math.abs(conn.getJID().getBareJID().hashCode())
						% numWorkerThreads).offer(item);

//      ret = queues.get(Math.abs(conn.getJID().getBareJID().hashCode()
//          % numQueues)).offer(item, packet.getPriority().ordinal());
			} else {

				// Otherwise per destination address
				// If the packet elemTo is set then used it, otherwise just packetTo:
				if (packet.getStanzaTo() != null) {
					ret = workerThreads.get(Math.abs(packet.getStanzaTo().getBareJID().hashCode())
							% numWorkerThreads).offer(item);

//        ret = queues.get(Math.abs(packet.getStanzaTo().hashCode() % numQueues)).offer(item,
//            packet.getPriority().ordinal());
				} else {
					ret = workerThreads.get(Math.abs(packet.getTo().hashCode())).offer(item);

//        ret = queues.get(Math.abs(packet.getTo().hashCode() % numQueues)).offer(item,
//            packet.getPriority().ordinal());
				}
			}
		} catch (Exception e) {

			// This should not happen, but just in case until we are sure all
			// cases are catched.
			// Otherwise per destination address
			// If the packet elemTo is set then used it, otherwise just packetTo:<commit_msg>Improved packets distribution among threads for not yet authenticated sessions - registration and authentication packets.<commit_after>/*
 * Tigase Jabber/XMPP Server
 * Copyright (C) 2004-2008 "<NAME>" <<EMAIL>>
 *
 * This program is free software: you can redistribute it and/or modify
 * it under the terms of the GNU General Public License as published by
 * the Free Software Foundation, version 3 of the License.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with this program. Look for COPYING file in the top folder.
 * If not, see http://www.gnu.org/licenses/.
 *
 * $Rev$
 * Last modified by $Author$
 * $Date$
 */

package tigase.util;

//~--- non-JDK imports --------------------------------------------------------

import tigase.server.Packet;

import tigase.xmpp.XMPPProcessorIfc;
import tigase.xmpp.XMPPResourceConnection;

//~--- JDK imports ------------------------------------------------------------

import java.util.ArrayList;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.logging.Level;
import java.util.logging.Logger;

//~--- classes ----------------------------------------------------------------

/**
 * Created: Apr 21, 2009 8:50:50 PM
 *
 * @param <E>
 * @author <a href="mailto:<EMAIL>"><NAME></a>
 * @version $Rev$
 */
public class ProcessingThreads<E extends WorkerThread> {
	private static final Logger log = Logger.getLogger(ProcessingThreads.class.getName());

	//~--- fields ---------------------------------------------------------------

	private long droppedPackets = 0;

	// private int maxQueueSize = 10000;
	private String name = null;

	// Packets are put in queues in such a way that all packets for the same
	// user end-up in the same queue. This is important in some cases as
	// packet processing order does matter in some cases, especially for
	// roster processing.
	// Therefore it is also recommended that there is a single thread for
	// each queue but we can ditribute load increasing number of queues.
	// private int numQueues = 2;
	private int numWorkerThreads = 1;
	private ArrayList<E> workerThreads = null;

	//~--- constructors ---------------------------------------------------------

//private ArrayList<PriorityQueueAbstract<QueueItem>> queues =
//  new ArrayList<PriorityQueueAbstract<QueueItem>>(1);

	/**
	 * Constructs ...
	 *
	 *
	 * @param worker
	 * @param numWorkerThreads
	 * @param maxQueueSize
	 * @param name
	 * @throws ClassNotFoundException
	 * @throws InstantiationException
	 * @throws IllegalAccessException
	 */
	@SuppressWarnings({ "unchecked" })
	public ProcessingThreads(E worker, int numWorkerThreads, int maxQueueSize, String name)
			throws ClassNotFoundException, InstantiationException, IllegalAccessException {

		// this.numQueues = numQueues;
		// this.maxQueueSize = maxQueueSize;
		this.numWorkerThreads = numWorkerThreads;
		workerThreads = new ArrayList<E>(numWorkerThreads);
		this.name = name;

//  for (int i = 0; i < numQueues; i++) {
		// LinkedBlockingQueue<QueueItem> queue = new LinkedBlockingQueue<QueueItem>(maxQueueSize);
//  queues.add(queue);
		for (int j = 0; j < numWorkerThreads; j++) {
			WorkerThread t = worker.getNewInstance();

			t.setQueueMaxSize(maxQueueSize);
			t.setDaemon(true);
			t.setName(name + " Queue Worker " + j);
			t.start();
			workerThreads.add((E) t);
		}

//  }
	}

	//~--- methods --------------------------------------------------------------

	/**
	 * Method description
	 *
	 *
	 * @param processor
	 * @param packet
	 * @param conn
	 *
	 * @return
	 */
	public boolean addItem(XMPPProcessorIfc processor, Packet packet, XMPPResourceConnection conn) {
		boolean ret = false;
		QueueItem item = new QueueItem(processor, packet, conn);

		try {
			if ((item.getConn() != null) && item.getConn().isAuthorized()) {

				// Queueing packets per user...
				ret = workerThreads.get(Math.abs(conn.getJID().getBareJID().hashCode())
						% numWorkerThreads).offer(item);

//      ret = queues.get(Math.abs(conn.getJID().getBareJID().hashCode()
//          % numQueues)).offer(item, packet.getPriority().ordinal());
			} else {
				if (packet.getPacketFrom() != null) {

					// Queueing packets per user's connection...
					ret = workerThreads.get(Math.abs(packet.getPacketFrom().hashCode())
							% numWorkerThreads).offer(item);
				} else {

					// Otherwise per destination address
					// If the packet elemTo is set then used it, otherwise just packetTo:
					if (packet.getStanzaTo() != null) {
						ret = workerThreads.get(Math.abs(packet.getStanzaTo().getBareJID().hashCode())
								% numWorkerThreads).offer(item);

//          ret = queues.get(Math.abs(packet.getStanzaTo().hashCode() % numQueues)).offer(item,
//              packet.getPriority().ordinal());
					} else {
						ret = workerThreads.get(Math.abs(packet.getTo().hashCode())).offer(item);

//          ret = queues.get(Math.abs(packet.getTo().hashCode() % numQueues)).offer(item,
//              packet.getPriority().ordinal());
					}
				}
			}
		} catch (Exception e) {

			// This should not happen, but just in case until we are sure all
			// cases are catched.
			// Otherwise per destination address
			// If the packet elemTo is set then used it, otherwise just packetTo:
{'fraction_non_alphanumeric': 0.1832254432873628, 'fraction_numerical': 0.005816680739281359, 'mean_word_length': 2.598919648885888, 'pattern_counts': {'":': 0, '<': 41, '<?xml version=': 0, '>': 37, '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': '6123604', 'n_tokens_mistral': 3324, 'n_tokens_neox': 2981, 'n_words': 898}