<commit_before>package com.programmerdan.minecraft.civspy;

import java.sql.PreparedStatement;
import java.util.Map.Entry;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedTransferQueue;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import java.util.logging.Level;
import java.util.logging.Logger;

import com.programmerdan.minecraft.civspy.database.Database;

/**
 * Roughly similar approach to Manager; stage() unwinds data to batch it up, then
 * other task(s) come along to deliver to the database.
 * 
 * @author ProgrammerDan
 *
 */
public class DataBatcher {
	private final Database db;
	private final Logger logger;

	private final AtomicLong inflowCount;
	private final AtomicLong outflowCount;
	private final AtomicInteger workerCount;
	
	/**
	 * Threadpool to handle scheduled tasks, liveness tester 
	 */
	private final ScheduledExecutorService scheduler;
	
	/**
	 * Threadpool to handle batch constructors
	 */
	private final ExecutorService batchExecutor;
	
	/**
	 * The queue for unspooled aggregations.
	 */
	private final LinkedTransferQueue<BatchLine> batchQueue;
	
	private boolean active = true;
	
	/**
	 * Maximum number of elements to put into a single batch.
	 */
	private final long maxBatchSize = 100l;
	/**
	 * Maximum number of milliseconds to wait on more elements to join an open batch.
	 */
	private final long maxBatchWait = 1000l;
	/**
	 * Effectively, max simultaneous connections. A single batch is run against a single connection.
	 */
	private final int maxExecutors = 5;
	
	public DataBatcher(final Database db, final Logger logger) {
		this.db = db;
		this.logger = logger;
		this.inflowCount = new AtomicLong(0l);
		this.outflowCount = new AtomicLong(0l);
		this.workerCount = new AtomicInteger(0);
		
		this.batchQueue = new LinkedTransferQueue<BatchLine>();
		
		this.scheduler = Executors.newScheduledThreadPool(1);
		
		this.startWatcher();
		
		this.batchExecutor = Executors.newFixedThreadPool(this.maxExecutors);
	}
	
	public void shutdown() {
		active = false;
		int delay = 0;
		while (!this.batchQueue.isEmpty() && delay < 600) {
			try {
				Thread.sleep(1000l);
			} catch(Exception e) {}
			delay ++;
			if (delay % 30 == 0) {
				this.logger.log(Level.INFO, "Waiting on batch queue workers to finish up, {0} seconds so far", delay);
			}
		}
		if (delay >= 600) {
			this.logger.log(Level.WARNING, "Giving up on waiting. DATA LOSS MAY OCCUR.");
		}
		batchQueue.clear();
		this.scheduler.shutdown();
		this.batchExecutor.shutdown();
	}
	
	/** 
	 * This task just keeps a worker running from time to time, just to make sure someone is watching
	 * the incoming queue and help moderate against inflow explosion.
	 */
	private void startWatcher() {
		scheduler.scheduleAtFixedRate(new Runnable() {
			<commit_msg>Add await termination to prevent most data loss of aggregates queued up.<commit_after>package com.programmerdan.minecraft.civspy;

import java.sql.PreparedStatement;
import java.util.Map.Entry;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedTransferQueue;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import java.util.logging.Level;
import java.util.logging.Logger;

import com.programmerdan.minecraft.civspy.database.Database;

/**
 * Roughly similar approach to Manager; stage() unwinds data to batch it up, then
 * other task(s) come along to deliver to the database.
 * 
 * @author ProgrammerDan
 *
 */
public class DataBatcher {
	private final Database db;
	private final Logger logger;

	private final AtomicLong inflowCount;
	private final AtomicLong outflowCount;
	private final AtomicInteger workerCount;
	
	/**
	 * Threadpool to handle scheduled tasks, liveness tester 
	 */
	private final ScheduledExecutorService scheduler;
	
	/**
	 * Threadpool to handle batch constructors
	 */
	private final ExecutorService batchExecutor;
	
	/**
	 * The queue for unspooled aggregations.
	 */
	private final LinkedTransferQueue<BatchLine> batchQueue;
	
	private boolean active = true;
	
	/**
	 * Maximum number of elements to put into a single batch.
	 */
	private final long maxBatchSize = 100l;
	/**
	 * Maximum number of milliseconds to wait on more elements to join an open batch.
	 */
	private final long maxBatchWait = 1000l;
	/**
	 * Effectively, max simultaneous connections. A single batch is run against a single connection.
	 */
	private final int maxExecutors = 5;
	
	public DataBatcher(final Database db, final Logger logger) {
		this.db = db;
		this.logger = logger;
		this.inflowCount = new AtomicLong(0l);
		this.outflowCount = new AtomicLong(0l);
		this.workerCount = new AtomicInteger(0);
		
		this.batchQueue = new LinkedTransferQueue<BatchLine>();
		
		this.scheduler = Executors.newScheduledThreadPool(1);
		
		this.startWatcher();
		
		this.batchExecutor = Executors.newFixedThreadPool(this.maxExecutors);
	}
	
	public void shutdown() {
		active = false;
		int delay = 0;
		while (!this.batchQueue.isEmpty() && delay < 600) {
			try {
				Thread.sleep(1000l);
			} catch(Exception e) {}
			delay ++;
			if (delay % 30 == 0) {
				this.logger.log(Level.INFO, "Waiting on batch queue workers to finish up, {0} seconds so far", delay);
			}
		}
		if (delay >= 600) {
			this.logger.log(Level.WARNING, "Giving up on waiting. DATA LOSS MAY OCCUR.");
		}
		batchQueue.clear();
		this.scheduler.shutdown();
		this.batchExecutor.shutdown();
		
		try {
			if (!this.batchExecutor.awaitTermination(60l, TimeUnit.SECONDS)) {
				this.logger.log(Level.WARNING, "Giving up on waiting for batch commit; DATA LOSS MAY HAVE OCCURRED.");
			}
		} catch (InterruptedException ie) {
			this.logger.log(Level.WARNING, "Was awaiting batch saving to finish, but was interrupted.", ie);
		}
	}
	
	/** 
	 * This task just keeps a worker running from time to time, just to make sure someone is watching
	 * the incoming queue and help moderate against inflow explosion.
	 */
	private void startWatcher() {
		scheduler.scheduleAtFixedRate(new Runnable() {
			
{'fraction_non_alphanumeric': 0.09711895910780668, 'fraction_numerical': 0.008674101610904586, 'mean_word_length': 4.080251770259638, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 9, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, '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': '3112791', 'n_tokens_mistral': 2104, 'n_tokens_neox': 1833, 'n_words': 582}