<commit_before>/***************************************************************************
 * Copyright 2014 Kieker Project (http://kieker-monitoring.net)
 *
 * 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 kieker.monitoring.writer;

import java.util.List;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;

import kieker.common.configuration.Configuration;
import kieker.common.logging.Log;
import kieker.common.logging.LogFactory;
import kieker.common.record.IMonitoringRecord;
import kieker.common.record.misc.RegistryRecord;

/**
 * @author <NAME>
 * 
 * @since 1.3
 */
public abstract class AbstractAsyncWriter extends AbstractMonitoringWriter {

	/** The name of the configuration determining the size of the queue of this writer. */
	public static final String CONFIG_QUEUESIZE = "QueueSize";
	/** The name of the configuration determining the behavior of this writer in case of a full queue. */
	public static final String CONFIG_BEHAVIOR = "QueueFullBehavior";
	/** The name of the configuration determining the maximal shutdown delay of this writer (in milliseconds). */
	public static final String CONFIG_SHUTDOWNDELAY = "MaxShutdownDelay";

	private static final Log LOG = LogFactory.getLog(AbstractAsyncWriter.class);

	// internal variables
	/** The queue containing the records to be written. */
	protected final BlockingQueue<IMonitoringRecord> blockingQueue;
	private final List<AbstractAsyncThread> workers = new CopyOnWriteArrayList<AbstractAsyncThread>();
	private final int queueFullBehavior;
	private final int maxShutdownDelay;
	private final AtomicLong missedRecords;

	/**
	 * This constructor initializes the writer based on the given configuration.
	 * 
	 * @param configuration
	 *            The configuration for this writer.
	 */
	protected AbstractAsyncWriter(final Configuration configuration) {
		super(configuration);
		final String prefix = this.getClass().getName() + ".";

		final int queueFullBehaviorTmp = configuration.getIntProperty(prefix + CONFIG_BEHAVIOR);
		if ((queueFullBehaviorTmp < 0) || (queueFullBehaviorTmp > 2)) {
			LOG.warn("Unknown value '" + queueFullBehaviorTmp + "' for " + prefix + CONFIG_BEHAVIOR + "; using default value 0");
			this.queueFullBehavior = 0;
		} else {
			this.queueFullBehavior = queueFullBehaviorTmp;
		}
		this.missedRecords = new AtomicLong(0);
		this.blockingQueue = new ArrayBlockingQueue<IMonitoringRecord>(configuration.getIntProperty(prefix + CONFIG_QUEUESIZE));
		this.maxShutdownDelay = configuration.getIntProperty(prefix + CONFIG_SHUTDOWNDELAY);
	}

	/**
	 * {@inheritDoc} Make sure that the three required properties always have default values!
	 */
	@Override
	protected Configuration getDefaultConfiguration() {
		final Configuration configuration = new Configuration(super.getDefaultConfiguration());
		final String prefix = this.getClass().getName() + "."; // can't use this.prefix, maybe uninitialized
		configuration.setProperty(prefix + CONFIG_QUEUESIZE, "10000");
		configuration.setProperty(prefix + CONFIG_BEHAVIOR, "0");
		configuration.setProperty(prefix + CONFIG_SHUTDOWNDELAY, "-1");
		return configuration;
	}

	/**
	 * This method must be called at the end of the child constructor!
	 * 
	 * @param worker
	 *            The new worker.
	 */
	protected final void addWorker(final AbstractAsyncThread worker) {
		this.workers.add(worker);
		worker.setDaemon(true); // might lead to inconsistent data due to harsh shutdown
		worker.start();
	}

	/**
	 * The framework ensures, that this method is called only once!
	 */
	@Override
	public final void terminate() {
		final CountDownLatch cdl = new CountDownLatch(this.workers.size());
		for (final AbstractAsyncThread worker : this.workers) {
			worker.initShutdown(cdl); // notify all workers
		}
		boolean finished = false;
		try {
			if (this.maxShutdownDelay > -1) {
				LOG.info("Shutting down writers, waiting at most " + this.maxShutdownDelay + " milliseconds.");
				finished = cdl.await(this.maxShutdownDelay, TimeUnit.MILLISECONDS);
			} else {
				LOG.info("Shutting down writers.");
				cdl.await();
				finished = true;
			}
		} catch (final InterruptedException ex) {
			// we should be able to ignore an interrupted wait (finished is still false)
		}
		if (finished) {
			LOG.info("Writer shutdown complete.");
		} else {
			LOG.info("Writer shutdown incomplete, " + cdl.getCount() + " worker(s) halted.");
		}
	}

	/**
	 * {@inheritDoc}
	 */
	@Override
	public final boolean newMonitoringRecord(final IMonitoringRecord monitoringRecord) {
		try {
			switch (this.queueFullBehavior) {
			case 1: // blocks when queue full
				boolean interrupted = false;
				for (int i = 0; i < 10; i++) { // drop out if more than 10 times interrupted
					try {
						this.blockingQueue.put(monitoringRecord);
						if (interrupted) {
							LOG.warn("Interupted when adding new monitoring record to queue. Tries: " + i);
							Thread.currentThread().interrupt(); // propagate interrupt
						}
						return true;
					} catch (final InterruptedException ignore) {
						interrupted = true;
					}
				}
				return false;
			case 2: // does nothing if queue is full
				if (!this.blockingQueue.offer(monitoringRecord)) {
					// RegistryRecords must always be placed in the queue
					if (monitoringRecord instanceof RegistryRecord) {
						try {
							this.blockingQueue.put(monitoringRecord);
						} catch (final InterruptedException ignore) {
							LOG.error("Interrupted while adding RegistryRecord. Monitorig log will be corrupted!");
							return false;
						}<commit_msg>Improve handling of interruped in waiting for space in writer queue<commit_after>/***************************************************************************
 * Copyright 2014 Kieker Project (http://kieker-monitoring.net)
 *
 * 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 kieker.monitoring.writer;

import java.util.List;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;

import kieker.common.configuration.Configuration;
import kieker.common.logging.Log;
import kieker.common.logging.LogFactory;
import kieker.common.record.IMonitoringRecord;
import kieker.common.record.misc.RegistryRecord;

/**
 * @author <NAME>
 * 
 * @since 1.3
 */
public abstract class AbstractAsyncWriter extends AbstractMonitoringWriter {

	/** The name of the configuration determining the size of the queue of this writer. */
	public static final String CONFIG_QUEUESIZE = "QueueSize";
	/** The name of the configuration determining the behavior of this writer in case of a full queue. */
	public static final String CONFIG_BEHAVIOR = "QueueFullBehavior";
	/** The name of the configuration determining the maximal shutdown delay of this writer (in milliseconds). */
	public static final String CONFIG_SHUTDOWNDELAY = "MaxShutdownDelay";

	private static final Log LOG = LogFactory.getLog(AbstractAsyncWriter.class);

	// internal variables
	/** The queue containing the records to be written. */
	protected final BlockingQueue<IMonitoringRecord> blockingQueue;
	private final List<AbstractAsyncThread> workers = new CopyOnWriteArrayList<AbstractAsyncThread>();
	private final int queueFullBehavior;
	private final int maxShutdownDelay;
	private final AtomicLong missedRecords;

	/**
	 * This constructor initializes the writer based on the given configuration.
	 * 
	 * @param configuration
	 *            The configuration for this writer.
	 */
	protected AbstractAsyncWriter(final Configuration configuration) {
		super(configuration);
		final String prefix = this.getClass().getName() + ".";

		final int queueFullBehaviorTmp = configuration.getIntProperty(prefix + CONFIG_BEHAVIOR);
		if ((queueFullBehaviorTmp < 0) || (queueFullBehaviorTmp > 2)) {
			LOG.warn("Unknown value '" + queueFullBehaviorTmp + "' for " + prefix + CONFIG_BEHAVIOR + "; using default value 0");
			this.queueFullBehavior = 0;
		} else {
			this.queueFullBehavior = queueFullBehaviorTmp;
		}
		this.missedRecords = new AtomicLong(0);
		this.blockingQueue = new ArrayBlockingQueue<IMonitoringRecord>(configuration.getIntProperty(prefix + CONFIG_QUEUESIZE));
		this.maxShutdownDelay = configuration.getIntProperty(prefix + CONFIG_SHUTDOWNDELAY);
	}

	/**
	 * {@inheritDoc} Make sure that the three required properties always have default values!
	 */
	@Override
	protected Configuration getDefaultConfiguration() {
		final Configuration configuration = new Configuration(super.getDefaultConfiguration());
		final String prefix = this.getClass().getName() + "."; // can't use this.prefix, maybe uninitialized
		configuration.setProperty(prefix + CONFIG_QUEUESIZE, "10000");
		configuration.setProperty(prefix + CONFIG_BEHAVIOR, "0");
		configuration.setProperty(prefix + CONFIG_SHUTDOWNDELAY, "-1");
		return configuration;
	}

	/**
	 * This method must be called at the end of the child constructor!
	 * 
	 * @param worker
	 *            The new worker.
	 */
	protected final void addWorker(final AbstractAsyncThread worker) {
		this.workers.add(worker);
		worker.setDaemon(true); // might lead to inconsistent data due to harsh shutdown
		worker.start();
	}

	/**
	 * The framework ensures, that this method is called only once!
	 */
	@Override
	public final void terminate() {
		final CountDownLatch cdl = new CountDownLatch(this.workers.size());
		for (final AbstractAsyncThread worker : this.workers) {
			worker.initShutdown(cdl); // notify all workers
		}
		boolean finished = false;
		try {
			if (this.maxShutdownDelay > -1) {
				LOG.info("Shutting down writers, waiting at most " + this.maxShutdownDelay + " milliseconds.");
				finished = cdl.await(this.maxShutdownDelay, TimeUnit.MILLISECONDS);
			} else {
				LOG.info("Shutting down writers.");
				cdl.await();
				finished = true;
			}
		} catch (final InterruptedException ex) {
			// we should be able to ignore an interrupted wait (finished is still false)
		}
		if (finished) {
			LOG.info("Writer shutdown complete.");
		} else {
			LOG.info("Writer shutdown incomplete, " + cdl.getCount() + " worker(s) halted.");
		}
	}

	/**
	 * {@inheritDoc}
	 */
	@Override
	public final boolean newMonitoringRecord(final IMonitoringRecord monitoringRecord) {
		try {
			switch (this.queueFullBehavior) {
			case 1: // blocks when queue full
				for (int i = 0; i < 10; i++) { // drop out if more than 10 times interrupted
					try {
						this.blockingQueue.put(monitoringRecord);
						return true;
					} catch (final InterruptedException ignore) {
						LOG.warn("Interupted when adding new monitoring record to queue. Try: " + i);
						Thread.currentThread().interrupt(); // propagate interrupt
					}
				}
				LOG.error("Failed to add new monitoring record to queue (Finally interruped while blocked).");
				return false;
			case 2: // does nothing if queue is full
				if (!this.blockingQueue.offer(monitoringRecord)) {
					// RegistryRecords must always be placed in the queue
					if (monitoringRecord instanceof RegistryRecord) {
						try {
							this.blockingQueue.put(monitoringRecord);
						} catch (final InterruptedException ignore) {
							LOG.error("Interrupted while adding RegistryRecord. Monitorig log will be corrupted!");
							return false;
						}
{'fraction_non_alphanumeric': 0.1168220637781086, 'fraction_numerical': 0.004701089085638173, 'mean_word_length': 4.188617886178862, 'pattern_counts': {'":': 0, '<': 17, '<?xml version=': 0, '>': 17, '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': '5527023', 'n_tokens_mistral': 3830, 'n_tokens_neox': 3327, 'n_words': 1101}