<commit_before>package net.lecousin.framework.concurrent;

import java.io.Closeable;
import java.lang.management.LockInfo;
import java.lang.management.ManagementFactory;
import java.lang.management.MonitorInfo;
import java.lang.management.ThreadInfo;
import java.lang.management.ThreadMXBean;
import java.util.LinkedList;
import java.util.concurrent.ThreadFactory;

import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;

import net.lecousin.framework.application.LCCore;
import net.lecousin.framework.util.DebugUtil;

/** Monitoring of multi-threading system to avoid dead tasks. */
@SuppressFBWarnings({"MS_CANNOT_BE_FINAL", "MS_SHOULD_BE_FINAL" })
public final class TaskMonitoring {

	public static boolean checkLocksOfBlockingTasks = false;
	
	public static int MONITORING_INTERVAL = 60 * 1000;
	public static int SECONDS_BEFORE_TO_PUT_TASK_ASIDE = 5 * 60;
	public static int SECONDS_BEFORE_KILL_TASK = 10 * 60;
	
	private static TaskMonitor monitor;
	
	static synchronized void start(ThreadFactory threadFactory) {
		if (monitor != null) return;
		monitor = new TaskMonitor();
		threadFactory.newThread(monitor).start();
		LCCore.get().toClose(monitor);
	}
	
	private static void append(StringBuilder s, MonitorInfo[] monitors) {
		if (monitors.length > 0) {
			s.append("\r\nLocked monitors:");
			for (int i = 0; i < monitors.length; ++i) {
				StackTraceElement trace = monitors[i].getLockedStackFrame();
				s.append("\r\n - ").append(trace.getClassName()).append('#')
					.append(trace.getMethodName()).append(':').append(trace.getLineNumber());
			}
		}
	}
	
	private static void append(StringBuilder s, LockInfo[] locks) {
		if (locks.length > 0) {
			s.append("\r\nLocked synchronizers:");
			for (int i = 0; i < locks.length; ++i) {
				s.append("\r\n - ").append(locks[i].toString());
			}
		}
	}
	
	private static void appendLocks(StringBuilder s, Thread t) {
		ThreadMXBean bean = ManagementFactory.getThreadMXBean();
		if (bean == null) return;
		ThreadInfo info = bean.getThreadInfo(new long[] { t.getId() }, true, true)[0];
		if (info == null) return;
		MonitorInfo[] monitors = info.getLockedMonitors();
		LockInfo[] locks = info.getLockedSynchronizers();
		append(s, monitors);
		append(s, locks);
	}

	private static class TaskMonitor implements Runnable, Closeable {
		
		private Object lock = new Object();
		private boolean closed = false;
		
		@Override
		public void run() {
			while (!closed) {
				synchronized (lock) {
					try { lock.wait(MONITORING_INTERVAL); }
					catch (InterruptedException e) { break; }
					if (closed)
						break;
				}
				LinkedList<TaskWorker> blocked = new LinkedList<>();
				for (TaskManager manager : Threading.getAllTaskManagers())
					if (manager instanceof FixedThreadTaskManager) {
						FixedThreadTaskManager m = (FixedThreadTaskManager)manager;
						check(m);
						blocked.addAll(m.getBlockedWorkers());
					} else {
						// TODO
					}
				checkBlockedWorkers(blocked);
			}
		}
		
		@Override
		public void close() {
			closed = true;
			synchronized (lock) {
				lock.notifyAll();
			}
		}
	}
	
	private static void check(FixedThreadTaskManager manager) {
		for (TaskWorker worker : manager.getAllActiveWorkers())
			check(worker);
	}
	
	private static void check(TaskWorker worker) {
		if (worker == null) return;
		long now = System.nanoTime();
		Task<?,?> task = worker.currentTask;
		if (task == null) return;
		long start = worker.currentTaskStart;
		long seconds = (now - start) / (1000000L * 1000);
		if (seconds < SECONDS_BEFORE_TO_PUT_TASK_ASIDE) return;
		if (seconds < SECONDS_BEFORE_KILL_TASK) {
			StringBuilder s = new StringBuilder(1024);
			s.append("Task ").append(task).append(" is running since ").append(seconds)
			 .append(" seconds ! put it aside and start a new thread, current stack:\r\n");
			DebugUtil.createStackTrace(s, worker.thread.getStackTrace());
			appendLocks(s, worker.thread);
			Threading.logger.warn(s.toString());
			worker.manager.putWorkerAside(worker);
			return;
		}
		StringBuilder s = new StringBuilder(1024);
		s.append("Task ").append(task).append(" is running since ").append(seconds)
		 .append(" seconds ! kill it! current stack:\r\n");
		DebugUtil.createStackTrace(s, worker.thread.getStackTrace());
		appendLocks(s, worker.thread);
		Threading.logger.error(s.toString());
		worker.manager.killWorker(worker);
	}
	
	private static void checkBlockedWorkers(LinkedList<TaskWorker> workers) {
		if (!checkLocksOfBlockingTasks) return;
		if (workers.isEmpty()) return;
		ThreadMXBean bean = ManagementFactory.getThreadMXBean();
		if (bean == null) return;
		long[] ids = new long[workers.size()];
		int i = 0;
		for (TaskWorker w : workers) ids[i++] = w.thread.getId();
		ThreadInfo[] info = bean.getThreadInfo(ids, true, true);
		for (ThreadInfo ti : info) {
			MonitorInfo[] monitors = ti.getLockedMonitors();
			LockInfo[] locks = ti.getLockedSynchronizers();
			if (monitors.length == 0 && locks.length == 0) continue;
			StringBuilder s = new StringBuilder(4096);
			s.append("TaskWorker is blocked while locking objects:\r\n");
			DebugUtil.createStackTrace(s, ti.getStackTrace());
			append(s, monitors);
			append(s, locks);
			Threading.logger.error(s.toString());
		}
	}<commit_msg>TaskMonitoring: do not log blocked worker if no current task<commit_after>package net.lecousin.framework.concurrent;

import java.io.Closeable;
import java.lang.management.ManagementFactory;
import java.lang.management.MonitorInfo;
import java.lang.management.ThreadInfo;
import java.lang.management.ThreadMXBean;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.concurrent.ThreadFactory;

import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
import net.lecousin.framework.application.LCCore;
import net.lecousin.framework.util.DebugUtil;

/** Monitoring of multi-threading system to avoid dead tasks. */
@SuppressFBWarnings({"MS_CANNOT_BE_FINAL", "MS_SHOULD_BE_FINAL" })
public final class TaskMonitoring {

	public static boolean checkLocksOfBlockingTasks = false;
	
	public static int MONITORING_INTERVAL = 60 * 1000;
	public static int SECONDS_BEFORE_TO_PUT_TASK_ASIDE = 5 * 60;
	public static int SECONDS_BEFORE_KILL_TASK = 10 * 60;
	
	private static TaskMonitor monitor;
	
	static synchronized void start(ThreadFactory threadFactory) {
		if (monitor != null) return;
		monitor = new TaskMonitor();
		threadFactory.newThread(monitor).start();
		LCCore.get().toClose(monitor);
	}
	
	private static void append(StringBuilder s, MonitorInfo[] monitors) {
		if (monitors.length > 0) {
			s.append("\r\nLocked monitors:");
			for (int i = 0; i < monitors.length; ++i) {
				StackTraceElement trace = monitors[i].getLockedStackFrame();
				s.append("\r\n - ").append(trace.getClassName()).append('#')
					.append(trace.getMethodName()).append(':').append(trace.getLineNumber());
			}
		}
	}
	
	private static void appendLocks(StringBuilder s, Thread t) {
		ThreadMXBean bean = ManagementFactory.getThreadMXBean();
		if (bean == null) return;
		ThreadInfo info = bean.getThreadInfo(new long[] { t.getId() }, true, false)[0];
		if (info == null) return;
		MonitorInfo[] monitors = info.getLockedMonitors();
		append(s, monitors);
	}

	private static class TaskMonitor implements Runnable, Closeable {
		
		private Object lock = new Object();
		private boolean closed = false;
		
		@Override
		public void run() {
			while (!closed) {
				synchronized (lock) {
					try { lock.wait(MONITORING_INTERVAL); }
					catch (InterruptedException e) { break; }
					if (closed)
						break;
				}
				LinkedList<TaskWorker> blocked = new LinkedList<>();
				for (TaskManager manager : Threading.getAllTaskManagers())
					if (manager instanceof FixedThreadTaskManager) {
						FixedThreadTaskManager m = (FixedThreadTaskManager)manager;
						check(m);
						blocked.addAll(m.getBlockedWorkers());
					} else {
						// TODO
					}
				checkBlockedWorkers(blocked);
			}
		}
		
		@Override
		public void close() {
			closed = true;
			synchronized (lock) {
				lock.notifyAll();
			}
		}
	}
	
	private static void check(FixedThreadTaskManager manager) {
		for (TaskWorker worker : manager.getAllActiveWorkers())
			check(worker);
	}
	
	private static void check(TaskWorker worker) {
		if (worker == null) return;
		long now = System.nanoTime();
		Task<?,?> task = worker.currentTask;
		if (task == null) return;
		long start = worker.currentTaskStart;
		long seconds = (now - start) / (1000000L * 1000);
		if (seconds < SECONDS_BEFORE_TO_PUT_TASK_ASIDE) return;
		if (seconds < SECONDS_BEFORE_KILL_TASK) {
			StringBuilder s = new StringBuilder(1024);
			s.append("Task ").append(task).append(" is running since ").append(seconds)
			 .append(" seconds ! put it aside and start a new thread, current stack:\r\n");
			DebugUtil.createStackTrace(s, worker.thread.getStackTrace());
			appendLocks(s, worker.thread);
			Threading.logger.warn(s.toString());
			worker.manager.putWorkerAside(worker);
			return;
		}
		StringBuilder s = new StringBuilder(1024);
		s.append("Task ").append(task).append(" is running since ").append(seconds)
		 .append(" seconds ! kill it! current stack:\r\n");
		DebugUtil.createStackTrace(s, worker.thread.getStackTrace());
		appendLocks(s, worker.thread);
		Threading.logger.error(s.toString());
		worker.manager.killWorker(worker);
	}
	
	private static void checkBlockedWorkers(LinkedList<TaskWorker> workers) {
		if (!checkLocksOfBlockingTasks) return;
		if (workers.isEmpty()) return;
		ThreadMXBean bean = ManagementFactory.getThreadMXBean();
		if (bean == null) return;
		long[] ids = new long[workers.size()];
		int i = 0;
		for (TaskWorker w : workers) ids[i++] = w.thread.getId();
		ThreadInfo[] info = bean.getThreadInfo(ids, true, false);
		Iterator<TaskWorker> it = workers.iterator();
		for (ThreadInfo ti : info) {
			TaskWorker w = it.next();
			Task<?,?> task = w.currentTask;
			if (task == null) continue;
			MonitorInfo[] monitors = ti.getLockedMonitors();
			if (monitors.length == 0) continue;
			StringBuilder s = new StringBuilder(1024);
			s.append("TaskWorker is blocked while locking objects in task ").append(task.description).append(":\r\n");
			DebugUtil.createStackTrace(s, ti.getStackTrace());
			append(s, monitors);
			Threading.logger.error(s.toString());
		}
	}
{'fraction_non_alphanumeric': 0.1305744888023369, 'fraction_numerical': 0.008276533592989289, 'mean_word_length': 4.324520476931053, 'pattern_counts': {'":': 1, '<': 20, '<?xml version=': 0, '>': 16, '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': '1894992', 'n_tokens_mistral': 3698, 'n_tokens_neox': 3190, 'n_words': 766}