From f7b473e109070298de76a39d7f3cae4a4d4c46d5 Mon Sep 17 00:00:00 2001
From: reusedname <155286845+reusedname@users.noreply.github.com>
Date: Sun, 10 Nov 2024 22:07:17 +0500
Subject: [PATCH 1/8] fix segment downloads choking
---
.../src/main/java/ctbrec/recorder/SimplifiedLocalRecorder.java | 3 ++-
1 file changed, 2 insertions(+), 1 deletion(-)
diff --git a/common/src/main/java/ctbrec/recorder/SimplifiedLocalRecorder.java b/common/src/main/java/ctbrec/recorder/SimplifiedLocalRecorder.java
index 52e80cfc..a2b60574 100644
--- a/common/src/main/java/ctbrec/recorder/SimplifiedLocalRecorder.java
+++ b/common/src/main/java/ctbrec/recorder/SimplifiedLocalRecorder.java
@@ -35,6 +35,7 @@ import static ctbrec.SubsequentAction.*;
import static ctbrec.event.Event.Type.MODEL_ONLINE;
import static java.lang.Thread.MAX_PRIORITY;
import static java.lang.Thread.MIN_PRIORITY;
+import static java.util.concurrent.TimeUnit.SECONDS;
@Slf4j
public class SimplifiedLocalRecorder implements Recorder {
@@ -56,7 +57,7 @@ public class SimplifiedLocalRecorder implements Recorder {
// thread pools for downloads and post-processing
private final ScheduledExecutorService scheduler;
private final ExecutorService playlistDownloadPool = Executors.newFixedThreadPool(10);
- private final ExecutorService segmentDownloadPool = Executors.newFixedThreadPool(10);
+ private final ExecutorService segmentDownloadPool = new ThreadPoolExecutor(0, 1000, 30L, SECONDS, new SynchronousQueue<>(), createThreadFactory("SegmentDownload", MAX_PRIORITY - 2));
private final ExecutorService postProcessing;
private final ThreadPoolScaler threadPoolScaler;
private long lastSpaceCheck;
From e6ccb9b66c0f7d4a091688d880ab950379fd6b1f Mon Sep 17 00:00:00 2001
From: reusedname <155286845+reusedname@users.noreply.github.com>
Date: Thu, 14 Nov 2024 23:16:42 +0500
Subject: [PATCH 2/8] basic debug servlet to print thread stacks
---
.../ctbrec/recorder/server/DebugServlet.java | 71 +++++++++++++++++++
.../ctbrec/recorder/server/HttpServer.java | 4 ++
2 files changed, 75 insertions(+)
create mode 100644 server/src/main/java/ctbrec/recorder/server/DebugServlet.java
diff --git a/server/src/main/java/ctbrec/recorder/server/DebugServlet.java b/server/src/main/java/ctbrec/recorder/server/DebugServlet.java
new file mode 100644
index 00000000..dfe776ee
--- /dev/null
+++ b/server/src/main/java/ctbrec/recorder/server/DebugServlet.java
@@ -0,0 +1,71 @@
+package ctbrec.recorder.server;
+
+import ctbrec.Config;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.json.JSONObject;
+
+import javax.servlet.http.HttpServletRequest;
+import javax.servlet.http.HttpServletResponse;
+
+import java.io.PrintWriter;
+import java.io.StringWriter;
+import java.net.URLDecoder;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+import java.text.MessageFormat;
+
+import static java.nio.charset.StandardCharsets.UTF_8;
+import static javax.servlet.http.HttpServletResponse.*;
+
+@Slf4j
+public class DebugServlet extends AbstractCtbrecServlet {
+
+ public static final String BASE_URL = "/debug";
+
+ private static final Pattern URL_PATTERN_DEBUG_STACK = Pattern.compile(BASE_URL + "/stack(/.*?)");
+
+ public DebugServlet()
+ {}
+
+ @Override
+ protected void doGet(HttpServletRequest req, HttpServletResponse resp) {
+ String requestURI = req.getRequestURI().substring(req.getContextPath().length());
+ try {
+ // boolean authenticated = checkAuthentication(req, "");
+ // if (!authenticated) {
+ // sendResponse(resp, SC_UNAUTHORIZED, HMAC_ERROR_DOCUMENT);
+ // return;
+ // }
+
+ Matcher m;
+ if ((m = URL_PATTERN_DEBUG_STACK.matcher(requestURI)).matches()) {
+ String threadUrl = URLDecoder.decode(m.group(1), UTF_8);
+
+ var stacks = Thread.getAllStackTraces();
+ var box = new Object() { String text = ""; };//stacks.toString();
+
+ StringWriter sw = new StringWriter();
+ PrintWriter pw = new PrintWriter(sw);
+ // e.printStackTrace(pw);
+
+ stacks.forEach((thread, stack) -> {
+ box.text += String.format("%s:\n", thread.getName());
+ var idx = 0;
+ for (var s : stack) {
+ box.text += String.format("[%d] %s\n", idx++, s.toString());
+ }
+ });
+
+ log.debug("Stacks Request {} - {}", threadUrl, stacks);
+ resp.setContentType("text/plain");
+ sendResponse(resp, SC_OK, box.text);
+ } else
+ sendResponse(resp, SC_NOT_FOUND, "");
+ } catch (Exception e) {
+ log.error(INTERNAL_SERVER_ERROR, e);
+ sendResponse(resp, SC_INTERNAL_SERVER_ERROR, INTERNAL_SERVER_ERROR);
+ }
+ }
+
+}
diff --git a/server/src/main/java/ctbrec/recorder/server/HttpServer.java b/server/src/main/java/ctbrec/recorder/server/HttpServer.java
index 7a4452ed..0b507503 100644
--- a/server/src/main/java/ctbrec/recorder/server/HttpServer.java
+++ b/server/src/main/java/ctbrec/recorder/server/HttpServer.java
@@ -258,6 +258,10 @@ public class HttpServer {
ModelServlet modelServlet = new ModelServlet(config);
holder = new ServletHolder(modelServlet);
defaultContext.addServlet(holder, ModelServlet.BASE_URL + "/*");
+
+ DebugServlet debugServlet = new DebugServlet();
+ holder = new ServletHolder(debugServlet);
+ defaultContext.addServlet(holder, DebugServlet.BASE_URL + "/*");
if (this.config.getSettings().webinterface) {
startWebInterface(defaultContext, basicAuthContext);
From a003f84e70f7af65213a82875e136562ebce71c6 Mon Sep 17 00:00:00 2001
From: reusedname <155286845+reusedname@users.noreply.github.com>
Date: Wed, 20 Nov 2024 18:53:36 +0500
Subject: [PATCH 3/8] use library redirects for ffmpeg output
---
.../src/main/java/ctbrec/recorder/FFmpeg.java | 75 +++++++------------
1 file changed, 25 insertions(+), 50 deletions(-)
diff --git a/common/src/main/java/ctbrec/recorder/FFmpeg.java b/common/src/main/java/ctbrec/recorder/FFmpeg.java
index b6bb9ebc..bd1e4a19 100644
--- a/common/src/main/java/ctbrec/recorder/FFmpeg.java
+++ b/common/src/main/java/ctbrec/recorder/FFmpeg.java
@@ -1,67 +1,43 @@
package ctbrec.recorder;
import java.io.File;
-import java.io.FileOutputStream;
import java.io.IOException;
-import java.io.OutputStream;
+import java.lang.ProcessBuilder.Redirect;
import java.nio.file.Files;
import java.util.Arrays;
-import java.util.concurrent.Executors;
-import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.ThreadFactory;
import java.util.function.Consumer;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
+import lombok.extern.slf4j.Slf4j;
-import ctbrec.io.DevNull;
-import ctbrec.io.ProcessStreamRedirector;
import ctbrec.recorder.download.ProcessExitedUncleanException;
+@Slf4j
public class FFmpeg {
-
- private static final Logger LOG = LoggerFactory.getLogger(FFmpeg.class);
-
- private static ScheduledExecutorService processOutputReader = Executors.newScheduledThreadPool(2, createThreadFactory("FFmpeg output stream reader"));
-
private Process process;
private boolean logOutput = false;
private Consumer<Process> startCallback;
private Consumer<Integer> exitCallback;
private File ffmpegLog = null;
- private OutputStream ffmpegLogStream;
- private ProcessStreamRedirector stdoutRedirector;
- private ProcessStreamRedirector stderrRedirector;
private FFmpeg() {}
- private static ThreadFactory createThreadFactory(String name) {
- return r -> {
- Thread t = new Thread(r);
- t.setName(name);
- t.setDaemon(true);
- t.setPriority(Thread.MIN_PRIORITY);
- return t;
- };
- }
-
public void exec(String[] cmdline, String[] env, File executionDir) throws IOException {
- LOG.trace("FFmpeg command line: {}", Arrays.toString(cmdline));
- process = Runtime.getRuntime().exec(cmdline, env, executionDir);
- afterStart();
+ log.trace("FFmpeg command line: {}", Arrays.toString(cmdline));
+ // process = Runtime.getRuntime().exec(cmdline, env, executionDir);
+
+ var builder = new ProcessBuilder(cmdline);
+ // builder.environment().();
+ builder.directory(executionDir);
+ setupLogging(builder);
+ process = builder.start();
+
+ notifyStartCallback(process);
}
- private void afterStart() throws IOException {
- notifyStartCallback(process);
- setupLogging();
- }
public void shutdown(int exitCode) throws IOException {
- LOG.trace("FFmpeg exit code was {}", exitCode);
- ffmpegLogStream.flush();
- ffmpegLogStream.close();
- stdoutRedirector.setKeepGoing(false);
- stderrRedirector.setKeepGoing(false);
+ process.destroy();
+ log.trace("FFmpeg exit code was {}", exitCode);
notifyExitCallback(exitCode);
if (exitCode != 1) {
if (ffmpegLog != null && ffmpegLog.exists()) {
@@ -72,28 +48,27 @@ public class FFmpeg {
}
}
- private void setupLogging() throws IOException {
+ private void setupLogging(ProcessBuilder builder) throws IOException {
if (logOutput) {
if (ffmpegLog == null) {
ffmpegLog = File.createTempFile("ffmpeg_", ".log");
}
- LOG.trace("Logging FFmpeg output to {}", ffmpegLog);
+ log.trace("Logging FFmpeg output to {}", ffmpegLog);
ffmpegLog.deleteOnExit();
- ffmpegLogStream = new FileOutputStream(ffmpegLog);
+
+ builder.redirectOutput(Redirect.to(ffmpegLog));
+ builder.redirectErrorStream(true);
} else {
- ffmpegLogStream = new DevNull();
+ builder.redirectOutput(Redirect.DISCARD);
+ builder.redirectError(Redirect.DISCARD);
}
- stdoutRedirector = new ProcessStreamRedirector(processOutputReader, process.getInputStream(), ffmpegLogStream);
- stderrRedirector = new ProcessStreamRedirector(processOutputReader, process.getErrorStream(), ffmpegLogStream);
- processOutputReader.submit(stdoutRedirector);
- processOutputReader.submit(stderrRedirector);
}
private void notifyStartCallback(Process process) {
try {
startCallback.accept(process);
} catch(Exception e) {
- LOG.error("Exception in onStart callback", e);
+ log.error("Exception in onStart callback", e);
}
}
@@ -101,7 +76,7 @@ public class FFmpeg {
try {
exitCallback.accept(exitCode);
} catch(Exception e) {
- LOG.error("Exception in onExit callback", e);
+ log.error("Exception in onExit callback", e);
}
}
@@ -110,7 +85,7 @@ public class FFmpeg {
try {
shutdown(exitCode);
} catch (IOException e) {
- LOG.error("Error while shutting down FFmpeg process", e);
+ log.error("Error while shutting down FFmpeg process", e);
}
return exitCode;
}
From 5bc8a6fdf98267ddff7a785a988166cee0d34bf9 Mon Sep 17 00:00:00 2001
From: reusedname <155286845+reusedname@users.noreply.github.com>
Date: Tue, 3 Dec 2024 14:24:01 +0500
Subject: [PATCH 4/8] debug statistics page
---
common/src/main/java/ctbrec/io/HttpClient.java | 1 +
.../java/ctbrec/recorder/server/DebugServlet.java | 13 ++++++++++++-
2 files changed, 13 insertions(+), 1 deletion(-)
diff --git a/common/src/main/java/ctbrec/io/HttpClient.java b/common/src/main/java/ctbrec/io/HttpClient.java
index e90a6b79..200983dd 100644
--- a/common/src/main/java/ctbrec/io/HttpClient.java
+++ b/common/src/main/java/ctbrec/io/HttpClient.java
@@ -37,6 +37,7 @@ import static java.nio.charset.StandardCharsets.UTF_8;
@Slf4j
public abstract class HttpClient {
+ @Getter
private static final ConnectionPool GLOBAL_HTTP_CONN_POOL = new ConnectionPool(10, 2, TimeUnit.MINUTES);
@Getter
diff --git a/server/src/main/java/ctbrec/recorder/server/DebugServlet.java b/server/src/main/java/ctbrec/recorder/server/DebugServlet.java
index dfe776ee..97eef9de 100644
--- a/server/src/main/java/ctbrec/recorder/server/DebugServlet.java
+++ b/server/src/main/java/ctbrec/recorder/server/DebugServlet.java
@@ -1,6 +1,7 @@
package ctbrec.recorder.server;
import ctbrec.Config;
+import ctbrec.io.HttpClient;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.json.JSONObject;
@@ -24,6 +25,7 @@ public class DebugServlet extends AbstractCtbrecServlet {
public static final String BASE_URL = "/debug";
private static final Pattern URL_PATTERN_DEBUG_STACK = Pattern.compile(BASE_URL + "/stack(/.*?)");
+ private static final Pattern URL_PATTERN_DEBUG_STATS = Pattern.compile(BASE_URL + "/stats");
public DebugServlet()
{}
@@ -57,9 +59,18 @@ public class DebugServlet extends AbstractCtbrecServlet {
}
});
- log.debug("Stacks Request {} - {}", threadUrl, stacks);
+ log.debug("Stacks Request {}", threadUrl);
resp.setContentType("text/plain");
sendResponse(resp, SC_OK, box.text);
+ } else if ((m = URL_PATTERN_DEBUG_STATS.matcher(requestURI)).matches()) {
+ String text = "";
+ text += String.format("GLOBAL_HTTP_CONN_POOL: connectionCount=%d, idleConnectionCount=%d\n",
+ HttpClient.getGLOBAL_HTTP_CONN_POOL().connectionCount(),
+ HttpClient.getGLOBAL_HTTP_CONN_POOL().idleConnectionCount());
+
+ log.debug("Stats Request");
+ resp.setContentType("text/plain");
+ sendResponse(resp, SC_OK, text);
} else
sendResponse(resp, SC_NOT_FOUND, "");
} catch (Exception e) {
From 435cbbdb95f174572c7fe481dd0aad6c2da24189 Mon Sep 17 00:00:00 2001
From: reusedname <155286845+reusedname@users.noreply.github.com>
Date: Tue, 17 Dec 2024 11:05:06 +0500
Subject: [PATCH 5/8] debug statistics (recording iteration lag)
---
common/pom.xml | 10 +++++++++
.../recorder/SimplifiedLocalRecorder.java | 1 +
.../main/java/ctbrec/recorder/Statistics.java | 21 +++++++++++++++++++
.../download/hls/AbstractHlsDownload.java | 2 ++
.../ctbrec/recorder/server/DebugServlet.java | 9 ++++++++
5 files changed, 43 insertions(+)
create mode 100644 common/src/main/java/ctbrec/recorder/Statistics.java
diff --git a/common/pom.xml b/common/pom.xml
index 226beee5..e7268b9b 100644
--- a/common/pom.xml
+++ b/common/pom.xml
@@ -88,6 +88,16 @@
<groupId>ch.qos.logback</groupId>
<artifactId>logback-classic</artifactId>
<scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>com.fasterxml.jackson.core</groupId>
+ <artifactId>jackson-databind</artifactId>
+ <version>2.10.0.pr1</version>
+ </dependency>
+ <dependency>
+ <groupId>commons-collections</groupId>
+ <artifactId>commons-collections</artifactId>
+ <version>3.2.2</version>
</dependency>
</dependencies>
diff --git a/common/src/main/java/ctbrec/recorder/SimplifiedLocalRecorder.java b/common/src/main/java/ctbrec/recorder/SimplifiedLocalRecorder.java
index a2b60574..d5ee142a 100644
--- a/common/src/main/java/ctbrec/recorder/SimplifiedLocalRecorder.java
+++ b/common/src/main/java/ctbrec/recorder/SimplifiedLocalRecorder.java
@@ -39,6 +39,7 @@ import static java.util.concurrent.TimeUnit.SECONDS;
@Slf4j
public class SimplifiedLocalRecorder implements Recorder {
+ public static final Statistics STATS = new Statistics();
public static final boolean IGNORE_CACHE = true;
private final List<Model> models = Collections.synchronizedList(new ArrayList<>());
diff --git a/common/src/main/java/ctbrec/recorder/Statistics.java b/common/src/main/java/ctbrec/recorder/Statistics.java
new file mode 100644
index 00000000..61583164
--- /dev/null
+++ b/common/src/main/java/ctbrec/recorder/Statistics.java
@@ -0,0 +1,21 @@
+package ctbrec.recorder;
+
+import java.util.concurrent.locks.ReadWriteLock;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+
+import org.apache.commons.collections.buffer.CircularFifoBuffer;
+
+
+public class Statistics {
+ public ReadWriteLock statsLock = new ReentrantReadWriteLock();
+ public CircularFifoBuffer stats = new CircularFifoBuffer(100);
+
+ public void add(Object val) {
+ statsLock.writeLock().lock();
+ try {
+ stats.add(val);
+ } finally {
+ statsLock.writeLock().unlock();
+ }
+ }
+}
diff --git a/common/src/main/java/ctbrec/recorder/download/hls/AbstractHlsDownload.java b/common/src/main/java/ctbrec/recorder/download/hls/AbstractHlsDownload.java
index 8c161f8d..80a205da 100644
--- a/common/src/main/java/ctbrec/recorder/download/hls/AbstractHlsDownload.java
+++ b/common/src/main/java/ctbrec/recorder/download/hls/AbstractHlsDownload.java
@@ -14,6 +14,7 @@ import ctbrec.io.HttpClient;
import ctbrec.io.HttpConstants;
import ctbrec.io.HttpException;
import ctbrec.recorder.InvalidPlaylistException;
+import ctbrec.recorder.SimplifiedLocalRecorder;
import ctbrec.recorder.download.AbstractDownload;
import ctbrec.recorder.download.HttpHeaderFactory;
import ctbrec.recorder.download.StreamSource;
@@ -108,6 +109,7 @@ public abstract class AbstractHlsDownload extends AbstractDownload {
@Override
public AbstractHlsDownload call() throws Exception {
+ SimplifiedLocalRecorder.STATS.add(Duration.between(rescheduleTime, Instant.now()).toMillis());
try {
if (segmentPlaylistUrl == null) {
diff --git a/server/src/main/java/ctbrec/recorder/server/DebugServlet.java b/server/src/main/java/ctbrec/recorder/server/DebugServlet.java
index 97eef9de..22a9cf47 100644
--- a/server/src/main/java/ctbrec/recorder/server/DebugServlet.java
+++ b/server/src/main/java/ctbrec/recorder/server/DebugServlet.java
@@ -2,6 +2,7 @@ package ctbrec.recorder.server;
import ctbrec.Config;
import ctbrec.io.HttpClient;
+import ctbrec.recorder.SimplifiedLocalRecorder;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.json.JSONObject;
@@ -68,6 +69,14 @@ public class DebugServlet extends AbstractCtbrecServlet {
HttpClient.getGLOBAL_HTTP_CONN_POOL().connectionCount(),
HttpClient.getGLOBAL_HTTP_CONN_POOL().idleConnectionCount());
+ var rlock = SimplifiedLocalRecorder.STATS.statsLock.readLock();
+ rlock.lock();
+ try {
+ text += String.format("Time between download iterations = %s\n", SimplifiedLocalRecorder.STATS.stats.toString());
+ } finally {
+ rlock.unlock();
+ }
+
log.debug("Stats Request");
resp.setContentType("text/plain");
sendResponse(resp, SC_OK, text);
From 74852692d07fb832751f2ad2de8aa7e3cb1c88ad Mon Sep 17 00:00:00 2001
From: reusedname <155286845+reusedname@users.noreply.github.com>
Date: Tue, 17 Dec 2024 21:31:07 +0500
Subject: [PATCH 6/8] virtual thread-based recording loops (one per recording)
(cherry picked from commit a4c44a4bc0a25e76715354bfa6e71681c0a8a50b)
---
.../recorder/SimplifiedLocalRecorder.java | 174 ++++++++++--------
1 file changed, 93 insertions(+), 81 deletions(-)
diff --git a/common/src/main/java/ctbrec/recorder/SimplifiedLocalRecorder.java b/common/src/main/java/ctbrec/recorder/SimplifiedLocalRecorder.java
index d5ee142a..067d9f22 100644
--- a/common/src/main/java/ctbrec/recorder/SimplifiedLocalRecorder.java
+++ b/common/src/main/java/ctbrec/recorder/SimplifiedLocalRecorder.java
@@ -1,6 +1,7 @@
package ctbrec.recorder;
import com.google.common.eventbus.Subscribe;
+
import ctbrec.*;
import ctbrec.Recording.State;
import ctbrec.event.*;
@@ -29,6 +30,7 @@ import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.locks.ReentrantLock;
import java.util.stream.Collectors;
+import com.google.common.collect.Ordering;
import static ctbrec.Recording.State.WAITING;
import static ctbrec.SubsequentAction.*;
@@ -40,7 +42,7 @@ import static java.util.concurrent.TimeUnit.SECONDS;
@Slf4j
public class SimplifiedLocalRecorder implements Recorder {
public static final Statistics STATS = new Statistics();
-
+
public static final boolean IGNORE_CACHE = true;
private final List<Model> models = Collections.synchronizedList(new ArrayList<>());
private final Config config;
@@ -53,28 +55,24 @@ public class SimplifiedLocalRecorder implements Recorder {
private final RecordingManager recordingManager;
private final RecordingPreconditions preconditions;
- private final BlockingQueue<Recording> recordings = new LinkedBlockingQueue<>();
-
+
// thread pools for downloads and post-processing
- private final ScheduledExecutorService scheduler;
- private final ExecutorService playlistDownloadPool = Executors.newFixedThreadPool(10);
- private final ExecutorService segmentDownloadPool = new ThreadPoolExecutor(0, 1000, 30L, SECONDS, new SynchronousQueue<>(), createThreadFactory("SegmentDownload", MAX_PRIORITY - 2));
- private final ExecutorService postProcessing;
- private final ThreadPoolScaler threadPoolScaler;
+ private final ExecutorService segmentDownloadPool = Executors.newVirtualThreadPerTaskExecutor();
+ private final ExecutorService recordingLoopPool = Executors.newVirtualThreadPerTaskExecutor();
+ private final ThreadPoolExecutor postProcessing;
+ private final Thread maintenanceThread;
private long lastSpaceCheck;
public SimplifiedLocalRecorder(Config config, List<Site> sites) throws IOException {
this.config = config;
client = new RecorderHttpClient(config);
- scheduler = Executors.newScheduledThreadPool(5, createThreadFactory("Download", MAX_PRIORITY));
- threadPoolScaler = new ThreadPoolScaler((ThreadPoolExecutor) scheduler, 5);
recordingManager = new RecordingManager(config, sites);
loadModels(sites);
int ppThreads = config.getSettings().postProcessingThreads;
BlockingQueue<Runnable> ppQueue = new LinkedBlockingQueue<>();
postProcessing = new ThreadPoolExecutor(ppThreads, ppThreads, 5, TimeUnit.MINUTES, ppQueue, createThreadFactory("PP", MIN_PRIORITY));
-
+
running = true;
registerEventBusListener();
@@ -84,21 +82,20 @@ public class SimplifiedLocalRecorder implements Recorder {
log.info("Models to record: {}", models);
log.info("Saving recordings in {}", config.getSettings().recordingsDir);
- startRecordingLoop();
+ maintenanceThread = startMaintenanceLoop();
}
- private void startRecordingLoop() {
- new Thread(() -> {
- while (running) {
- Recording rec = recordings.poll();
- if (rec != null) {
- processRecording(rec);
- }
+ private Thread startMaintenanceLoop() {
+ var t = new Thread(() -> {
+ while (running && !Thread.currentThread().isInterrupted()) {
checkFreeSpace();
- threadPoolScaler.tick();
- waitABit(100);
+ //threadPoolScaler.tick();
+ waitABit(1000);
}
- }).start();
+ });
+ t.setName("Recording loop");
+ t.start();
+ return t;
}
private void checkFreeSpace() {
@@ -125,33 +122,42 @@ public class SimplifiedLocalRecorder implements Recorder {
}
}
- private void processRecording(Recording recording) {
- if (recording.getCurrentIteration().isDone()) {
- if (recording.getRecordingProcess().isRunning()) {
- try {
- Instant rescheduleAt = recording.getCurrentIteration().get().getRescheduleTime();
- Duration duration = Duration.between(Instant.now(), rescheduleAt);
- long delayInMillis = Math.max(0, duration.toMillis());
- log.trace("Current iteration is done {}. Recording status {}. Rescheduling in {}ms", recording.getModel().getName(), recording.getStatus().name(), delayInMillis);
- scheduleRecording(recording, delayInMillis);
- } catch (InterruptedException e) {
- Thread.currentThread().interrupt();
- fail(recording);
- } catch (ExecutionException e) {
- log.error("Error while recording model {}. Stopping recording.", recording.getModel(), e);
- fail(recording);
+ private void singleRecordingLoop(Recording recording) {
+ while (recording.getRecordingProcess().isRunning()) {
+ try {
+ // run single iteration
+ recording.getRecordingProcess().call();
+
+ if (recording.getModel().isSuspended()) {
+ log.info("Recording process for suspended model found: {}. Stopping now", recording.getModel());
+ stopRecordingProcess(recording);
+ submitPostProcessingJob(recording);
+ break;
}
- } else {
- removeRecordingProcess(recording);
- if (deleteIfEmpty(recording)) {
- return;
- }
- submitPostProcessingJob(recording);
- tryRestartRecording(recording.getModel());
+
+ // wait necessary time
+ Instant rescheduleAt = recording.getRecordingProcess().getRescheduleTime();
+ Duration duration = Duration.between(Instant.now(), rescheduleAt);
+ long delayInMillis = Math.max(0, duration.toMillis());
+ log.trace("Current iteration is done {}. Recording status {}. Rescheduling in {}ms", recording.getModel().getName(), recording.getStatus().name(), delayInMillis);
+ Thread.sleep(delayInMillis);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ fail(recording);
+ return;
+ } catch (Exception e) {
+ log.error("Error while recording model {}. Stopping recording.", recording.getModel(), e);
+ fail(recording);
+ return;
}
- } else {
- recordings.add(recording);
+ }
+
+ removeRecordingProcess(recording);
+ if (deleteIfEmpty(recording)) {
+ return;
}
+ submitPostProcessingJob(recording);
+ tryRestartRecording(recording.getModel());
}
private void removeRecordingProcess(Recording rec) {
@@ -173,19 +179,6 @@ public class SimplifiedLocalRecorder implements Recorder {
tryRestartRecording(recording.getModel());
}
- private void scheduleRecording(Recording recording, long delayInMillis) {
- if (recording.getModel().isSuspended()) {
- log.info("Recording process for suspended model found: {}. Stopping now", recording.getModel());
- stopRecordingProcess(recording);
- submitPostProcessingJob(recording);
- return;
- }
- ScheduledFuture<RecordingProcess> future = scheduler.schedule(recording.getRecordingProcess(), delayInMillis, TimeUnit.MILLISECONDS);
- recording.setCurrentIteration(future);
- recording.getSelectedResolution();
- recordings.add(recording);
- }
-
private void loadModels(List<Site> sites) {
config.getSettings().models
.stream()
@@ -217,9 +210,16 @@ public class SimplifiedLocalRecorder implements Recorder {
private void stopRecordings() {
log.info("Stopping all recordings");
- for (Recording recording : recordings) {
- recording.getRecordingProcess().stop();
- recording.getRecordingProcess().awaitEnd();
+ recorderLock.lock();
+ try {
+ for (Recording recording : recordingProcesses) {
+ recording.getRecordingProcess().stop();
+ }
+ for (Recording recording : recordingProcesses) {
+ recording.getRecordingProcess().awaitEnd();
+ }
+ } finally {
+ recorderLock.unlock();
}
waitForRecordingsToTerminate();
log.info("Recordings have been stopped");
@@ -518,11 +518,11 @@ public class SimplifiedLocalRecorder implements Recorder {
public void shutdown(boolean immediately) {
log.info("Shutting down");
shuttingDown = true;
+ maintenanceThread.interrupt();
if (!immediately) {
try {
stopRecordings();
- shutdownPool("Scheduler", scheduler, 60);
- shutdownPool("PlaylistDownloadPool", playlistDownloadPool, 60);
+ shutdownPool("Recording loops", recordingLoopPool, 60);
shutdownPool("SegmentDownloadPool", segmentDownloadPool, 60);
shutdownPool("Post-Processing", postProcessing, 600);
} catch (InterruptedException e) {
@@ -733,18 +733,30 @@ public class SimplifiedLocalRecorder implements Recorder {
return;
}
- try {
- boolean modelInRecordingList = isTracked(model);
- boolean online = model.isOnline(IGNORE_CACHE);
- if (modelInRecordingList && online) {
- log.info("Restarting recording for model {}", model);
- startRecordingProcess(model);
- }
- } catch (InterruptedException e) {
- Thread.currentThread().interrupt();
- log.error("Couldn't restart recording for model {}", model);
- } catch (Exception e) {
- log.error("Couldn't restart recording for model {}", model);
+ boolean modelInRecordingList = isTracked(model);
+
+ if (modelInRecordingList) {
+ // .isOnline() check does blocking http request, so do this async
+ recordingLoopPool.submit(() -> {
+ try {
+ boolean online = model.isOnline(IGNORE_CACHE);
+ if (online) {
+ log.info("Restarting recording for model {}", model);
+
+ try {
+ recorderLock.lock();
+ startRecordingProcess(model);
+ } finally {
+ recorderLock.unlock();
+ }
+ }
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ log.error("Couldn't restart recording for model {}", model);
+ } catch (Exception e) {
+ log.error("Couldn't restart recording for model {}", model);
+ }
+ });
}
}
@@ -772,8 +784,8 @@ public class SimplifiedLocalRecorder implements Recorder {
});
}
- private CompletableFuture<Void> startRecordingProcess(Model model) {
- return CompletableFuture.runAsync(() -> {
+ private void startRecordingProcess(Model model) {
+ recordingLoopPool.submit(() -> {
recorderLock.lock();
try {
preconditions.check(model);
@@ -783,7 +795,7 @@ public class SimplifiedLocalRecorder implements Recorder {
setRecordingStatus(rec, State.RECORDING);
rec.getModel().setLastRecorded(rec.getStartDate());
recordingManager.saveRecording(rec);
- scheduleRecording(rec, 0);
+ recordingLoopPool.submit(() -> {singleRecordingLoop(rec);});
} catch (RecordUntilExpiredException e) {
log.info("Precondition not met to record {}: {}", model, e.getLocalizedMessage());
executeRecordUntilSubsequentAction(model);
@@ -794,12 +806,12 @@ public class SimplifiedLocalRecorder implements Recorder {
} finally {
recorderLock.unlock();
}
- }, segmentDownloadPool);
+ });
}
private ThreadFactory createThreadFactory(String name, int priority) {
return r -> {
- Thread t = new Thread(r);
+ Thread t = Thread.ofPlatform().unstarted(r);
t.setName(name + " " + UUID.randomUUID().toString().substring(0, 8));
t.setDaemon(true);
t.setPriority(priority);
From af38886f9739ce9e533315d3aeda4f51a2ae93ad Mon Sep 17 00:00:00 2001
From: reusedname <155286845+reusedname@users.noreply.github.com>
Date: Sun, 19 Jan 2025 11:35:56 +0500
Subject: [PATCH 7/8] faster shutdown, adjust download reschedule time
(cherry picked from commit ef818843185eba80fbe58786c91e2bc52994f9da)
---
.../main/java/ctbrec/recorder/SimplifiedLocalRecorder.java | 2 +-
.../ctbrec/recorder/download/hls/AbstractHlsDownload.java | 4 +++-
2 files changed, 4 insertions(+), 2 deletions(-)
diff --git a/common/src/main/java/ctbrec/recorder/SimplifiedLocalRecorder.java b/common/src/main/java/ctbrec/recorder/SimplifiedLocalRecorder.java
index 067d9f22..d5383409 100644
--- a/common/src/main/java/ctbrec/recorder/SimplifiedLocalRecorder.java
+++ b/common/src/main/java/ctbrec/recorder/SimplifiedLocalRecorder.java
@@ -728,7 +728,7 @@ public class SimplifiedLocalRecorder implements Recorder {
}
private void tryRestartRecording(Model model) {
- if (!running) {
+ if (!running || shuttingDown) {
// recorder is not in recording state
return;
}
diff --git a/common/src/main/java/ctbrec/recorder/download/hls/AbstractHlsDownload.java b/common/src/main/java/ctbrec/recorder/download/hls/AbstractHlsDownload.java
index 80a205da..c09f5ab7 100644
--- a/common/src/main/java/ctbrec/recorder/download/hls/AbstractHlsDownload.java
+++ b/common/src/main/java/ctbrec/recorder/download/hls/AbstractHlsDownload.java
@@ -154,7 +154,7 @@ public abstract class AbstractHlsDownload extends AbstractDownload {
model.delay();
stop();
} else {
- rescheduleTime = beforeLastPlaylistRequest; // try again as fast as possible
+ rescheduleTime = Instant.now(); // try again as fast as possible
}
} catch (EOFException e) {
// end of playlist reached
@@ -422,6 +422,8 @@ public abstract class AbstractHlsDownload extends AbstractDownload {
private void calculateRescheduleTime() {
rescheduleTime = beforeLastPlaylistRequest.plusMillis(1000);
+ if (Instant.now().isAfter(rescheduleTime))
+ rescheduleTime = Instant.now();
recordingEvents.add(RecordingEvent.of("next playlist download scheduled for " + rescheduleTime.toString()));
}
From 4dbcbaa7078a71cacf90cd0165045958ced8302a Mon Sep 17 00:00:00 2001
From: reusedname <155286845+reusedname@users.noreply.github.com>
Date: Thu, 6 Feb 2025 18:22:57 +0500
Subject: [PATCH 8/8] more debug stats
(cherry picked from commit 7f06a41dc081f45683218aaaa0c3621040b5d0a0)
---
.../recorder/download/RecordingProcess.java | 4 +++
.../download/hls/MergedFfmpegHlsDownload.java | 8 ++++++
.../ctbrec/recorder/server/DebugServlet.java | 25 ++++++++++++++++---
.../ctbrec/recorder/server/HttpServer.java | 2 +-
4 files changed, 34 insertions(+), 5 deletions(-)
diff --git a/common/src/main/java/ctbrec/recorder/download/RecordingProcess.java b/common/src/main/java/ctbrec/recorder/download/RecordingProcess.java
index f502a6b9..30ee8612 100644
--- a/common/src/main/java/ctbrec/recorder/download/RecordingProcess.java
+++ b/common/src/main/java/ctbrec/recorder/download/RecordingProcess.java
@@ -62,4 +62,8 @@ public interface RecordingProcess extends Callable<RecordingProcess> {
void awaitEnd();
AtomicLong getDownloadedBytes();
+
+ default String getStats() {
+ return "";
+ }
}
diff --git a/common/src/main/java/ctbrec/recorder/download/hls/MergedFfmpegHlsDownload.java b/common/src/main/java/ctbrec/recorder/download/hls/MergedFfmpegHlsDownload.java
index 89a206f2..ceedd7cf 100644
--- a/common/src/main/java/ctbrec/recorder/download/hls/MergedFfmpegHlsDownload.java
+++ b/common/src/main/java/ctbrec/recorder/download/hls/MergedFfmpegHlsDownload.java
@@ -32,6 +32,14 @@ public class MergedFfmpegHlsDownload extends AbstractHlsDownload {
protected OutputStream ffmpegStdIn;
protected BlockingQueue<Future<SegmentDownload>> queue = new LinkedBlockingQueue<>();
protected Lock ffmpegStreamLock = new ReentrantLock();
+
+ public String getStats() {
+ String text = (running ? "RUN" : "stp") + String.format(" %d: ", queue.size());
+ for (var elem : queue) {
+ text += elem.isDone() ? "|" : "-";
+ }
+ return text;
+ }
public MergedFfmpegHlsDownload(HttpClient client) {
super(client);
diff --git a/server/src/main/java/ctbrec/recorder/server/DebugServlet.java b/server/src/main/java/ctbrec/recorder/server/DebugServlet.java
index 22a9cf47..d0a069b3 100644
--- a/server/src/main/java/ctbrec/recorder/server/DebugServlet.java
+++ b/server/src/main/java/ctbrec/recorder/server/DebugServlet.java
@@ -2,6 +2,7 @@ package ctbrec.recorder.server;
import ctbrec.Config;
import ctbrec.io.HttpClient;
+import ctbrec.recorder.Recorder;
import ctbrec.recorder.SimplifiedLocalRecorder;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
@@ -28,8 +29,12 @@ public class DebugServlet extends AbstractCtbrecServlet {
private static final Pattern URL_PATTERN_DEBUG_STACK = Pattern.compile(BASE_URL + "/stack(/.*?)");
private static final Pattern URL_PATTERN_DEBUG_STATS = Pattern.compile(BASE_URL + "/stats");
- public DebugServlet()
- {}
+ protected Recorder recorder;
+
+ public DebugServlet(Recorder rec)
+ {
+ this.recorder = rec;
+ }
@Override
protected void doGet(HttpServletRequest req, HttpServletResponse resp) {
@@ -64,7 +69,7 @@ public class DebugServlet extends AbstractCtbrecServlet {
resp.setContentType("text/plain");
sendResponse(resp, SC_OK, box.text);
} else if ((m = URL_PATTERN_DEBUG_STATS.matcher(requestURI)).matches()) {
- String text = "";
+ String text = "<html><body style=\"body { font-family: Monospace; }\">";
text += String.format("GLOBAL_HTTP_CONN_POOL: connectionCount=%d, idleConnectionCount=%d\n",
HttpClient.getGLOBAL_HTTP_CONN_POOL().connectionCount(),
HttpClient.getGLOBAL_HTTP_CONN_POOL().idleConnectionCount());
@@ -77,8 +82,20 @@ public class DebugServlet extends AbstractCtbrecServlet {
rlock.unlock();
}
+ text += "\nRecording stats:\n<table>";
+ text = text.replace("\n", "<br>");
+
+ for (var rec : recorder.getRecordings()) {
+ var proc = rec.getRecordingProcess();
+ if (proc == null) continue;
+ text += String.format("<tr><td>%s</td><td>%s</td></tr>\n", proc.getModel().getDisplayName(), proc.getStats());
+ }
+
+ text += "</table>";
+ text += "</body></html>";
+
log.debug("Stats Request");
- resp.setContentType("text/plain");
+ resp.setContentType("text/html");
sendResponse(resp, SC_OK, text);
} else
sendResponse(resp, SC_NOT_FOUND, "");
diff --git a/server/src/main/java/ctbrec/recorder/server/HttpServer.java b/server/src/main/java/ctbrec/recorder/server/HttpServer.java
index 0b507503..4f9e26af 100644
--- a/server/src/main/java/ctbrec/recorder/server/HttpServer.java
+++ b/server/src/main/java/ctbrec/recorder/server/HttpServer.java
@@ -259,7 +259,7 @@ public class HttpServer {
holder = new ServletHolder(modelServlet);
defaultContext.addServlet(holder, ModelServlet.BASE_URL + "/*");
- DebugServlet debugServlet = new DebugServlet();
+ DebugServlet debugServlet = new DebugServlet(recorder);
holder = new ServletHolder(debugServlet);
defaultContext.addServlet(holder, DebugServlet.BASE_URL + "/*");