1 parent
febd68973b
commit
594889ea10
6 files changed
+60
-19
No files matched your search
@@ -2,14 +2,13 @@
|
||||
|
||||
const fs = require('fs');
|
||||
|
||||
const COPY_BUFFER_SIZE = 4 * 1024 * 1024;
|
||||
|
||||
const copyWithProgress = async ({
|
||||
source,
|
||||
destination,
|
||||
jobLog,
|
||||
updateProgress,
|
||||
label,
|
||||
copyMethod = 'auto',
|
||||
maxRetries = 2,
|
||||
retryDelayMs = 1000,
|
||||
}) => {
|
||||
@@ -27,16 +26,12 @@ const copyWithProgress = async ({
|
||||
if (typeof jobLog === 'function') jobLog(`${message}\n`);
|
||||
};
|
||||
|
||||
const copyOnce = () => new Promise((resolve, reject) => {
|
||||
const copyOnceWithStream = () => new Promise((resolve, reject) => {
|
||||
const startMs = Date.now();
|
||||
let copiedBytes = 0;
|
||||
let nextProgressPercent = 0;
|
||||
// Larger buffers avoid issuing thousands of small writes when the
|
||||
// destination is an SMB/CIFS mount with noticeable round-trip latency.
|
||||
const rd = fs.createReadStream(source, { highWaterMark: COPY_BUFFER_SIZE });
|
||||
const wr = fs.createWriteStream(destination, {
|
||||
highWaterMark: COPY_BUFFER_SIZE,
|
||||
});
|
||||
const rd = fs.createReadStream(source, { highWaterMark: 4 * 1024 * 1024 });
|
||||
const wr = fs.createWriteStream(destination, { highWaterMark: 4 * 1024 * 1024 });
|
||||
let settled = false;
|
||||
const fail = (error) => {
|
||||
if (settled) return;
|
||||
@@ -45,7 +40,6 @@ const copyWithProgress = async ({
|
||||
wr.destroy();
|
||||
reject(error);
|
||||
};
|
||||
|
||||
rd.on('error', fail);
|
||||
wr.on('error', fail);
|
||||
wr.on('finish', () => {
|
||||
@@ -57,20 +51,45 @@ const copyWithProgress = async ({
|
||||
copiedBytes += chunk.length;
|
||||
const percent = sourceSize === 0 ? 100 : Math.floor((copiedBytes / sourceSize) * 100);
|
||||
if (percent < nextProgressPercent) return;
|
||||
|
||||
const elapsedSeconds = Math.max((Date.now() - startMs) / 1000, 0.1);
|
||||
const copiedMb = (copiedBytes / 1024 / 1024).toFixed(1);
|
||||
const speedMbPerSecond = (copiedBytes / 1024 / 1024 / elapsedSeconds).toFixed(1);
|
||||
report(
|
||||
percent,
|
||||
`[${label}] Progress: ${Math.min(percent, 100)}% `
|
||||
+ `(${copiedMb}/${sourceSizeMb} MB, ${speedMbPerSecond} MB/s)`,
|
||||
+ `(${(copiedBytes / 1024 / 1024).toFixed(1)}/${sourceSizeMb} MB, `
|
||||
+ `${(copiedBytes / 1024 / 1024 / elapsedSeconds).toFixed(1)} MB/s)`,
|
||||
);
|
||||
nextProgressPercent = (Math.floor(percent / 10) + 1) * 10;
|
||||
});
|
||||
rd.pipe(wr);
|
||||
});
|
||||
|
||||
const copyOnceWithNative = () => new Promise((resolve, reject) => {
|
||||
const startMs = Date.now();
|
||||
let settled = false;
|
||||
const progressTimer = setInterval(() => {
|
||||
if (settled || !fs.existsSync(destination)) return;
|
||||
const copiedBytes = fs.statSync(destination).size;
|
||||
const elapsedSeconds = Math.max((Date.now() - startMs) / 1000, 0.1);
|
||||
report(
|
||||
Math.floor((copiedBytes / sourceSize) * 100),
|
||||
`[${label}] Progress: ${Math.min(Math.floor((copiedBytes / sourceSize) * 100), 100)}% `
|
||||
+ `(${(copiedBytes / 1024 / 1024).toFixed(1)}/${sourceSizeMb} MB, `
|
||||
+ `${(copiedBytes / 1024 / 1024 / elapsedSeconds).toFixed(1)} MB/s)`,
|
||||
);
|
||||
}, 1000);
|
||||
fs.promises.copyFile(source, destination).then(() => {
|
||||
settled = true;
|
||||
clearInterval(progressTimer);
|
||||
resolve(((Date.now() - startMs) / 1000).toFixed(1));
|
||||
}).catch((error) => {
|
||||
settled = true;
|
||||
clearInterval(progressTimer);
|
||||
reject(error);
|
||||
});
|
||||
});
|
||||
|
||||
const copyOnce = copyMethod === 'stream' ? copyOnceWithStream : copyOnceWithNative;
|
||||
|
||||
let lastError;
|
||||
for (let attempt = 0; attempt <= maxRetries; attempt += 1) {
|
||||
if (attempt > 0) {
|
||||
|
||||
@@ -25,6 +25,14 @@ const details = () => ({
|
||||
inputUI: { type: 'dropdown', options: ['true', 'false'] },
|
||||
tooltip: 'Skip when the source is already inside workDir.',
|
||||
},
|
||||
{
|
||||
label: 'Copy method',
|
||||
name: 'copyMethod',
|
||||
type: 'string',
|
||||
defaultValue: 'auto',
|
||||
inputUI: { type: 'dropdown', options: ['auto', 'native', 'stream'] },
|
||||
tooltip: 'Auto uses the operating system copy API; stream is a fallback for problematic mounts.',
|
||||
},
|
||||
],
|
||||
outputs: [{ number: 1, tooltip: 'File is ready in workDir — continue to next plugin' }],
|
||||
});
|
||||
@@ -57,6 +65,7 @@ const plugin = async (args) => {
|
||||
jobLog,
|
||||
updateProgress: args.updateProgress,
|
||||
label: 'copyToWorkDir',
|
||||
copyMethod: args.inputs.copyMethod || 'auto',
|
||||
});
|
||||
|
||||
return {
|
||||
|
||||
@@ -25,6 +25,14 @@ const details = () => ({
|
||||
inputUI: { type: 'dropdown', options: ['true', 'false'] },
|
||||
tooltip: 'Delete the processed file after it has replaced the original.',
|
||||
},
|
||||
{
|
||||
label: 'Copy method',
|
||||
name: 'copyMethod',
|
||||
type: 'string',
|
||||
defaultValue: 'auto',
|
||||
inputUI: { type: 'dropdown', options: ['auto', 'native', 'stream'] },
|
||||
tooltip: 'Auto uses the operating system copy API; stream is a fallback for problematic mounts.',
|
||||
},
|
||||
],
|
||||
outputs: [{ number: 1, tooltip: 'Original file replaced — continue to next plugin' }],
|
||||
});
|
||||
@@ -62,6 +70,7 @@ const plugin = async (args) => {
|
||||
jobLog,
|
||||
updateProgress: args.updateProgress,
|
||||
label: 'replaceOriginalFile',
|
||||
copyMethod: args.inputs.copyMethod || 'auto',
|
||||
});
|
||||
|
||||
if (fs.existsSync(destination)) {
|
||||
|
||||
@@ -279,8 +279,9 @@ Performs 2-pass libx265 encoding using VBV parameters from `args.variables`.
|
||||
|
||||
Streams the source file to the worker's local `workDir`, verifies the byte count,
|
||||
reports throttled copy progress, and retries failed writes twice after removing
|
||||
the partial destination. It uses large buffered streams for better throughput
|
||||
when the library is an SMB/CIFS mount. It stores the original library path in
|
||||
the partial destination. Its `Copy method` input defaults to `auto` (the
|
||||
operating system's native file-copy path) and offers `stream` as a fallback
|
||||
for problematic mounts. It stores the original library path in
|
||||
`variables.originalLibraryFile` for `replaceOriginalFile`.
|
||||
|
||||
### 9. `replaceOriginalFile` — Replace Original File
|
||||
|
||||
@@ -122,8 +122,9 @@ Aggiungi al PATH di sistema:
|
||||
a network share. Place it before the first analysis/transcode node; it records
|
||||
the library path for `replaceOriginalFile` 2.0.0.
|
||||
Both progress-enabled copy operations verify the destination size and retry
|
||||
failed writes twice after removing the partial destination. They use larger
|
||||
buffers to avoid slow small-write behavior on SMB/CIFS mounts.
|
||||
failed writes twice after removing the partial destination. Their `Copy method`
|
||||
input defaults to `auto` (the operating system's native file-copy path), with
|
||||
`stream` available as a fallback for problematic SMB/CIFS mounts.
|
||||
Both progress-enabled plugins require
|
||||
`FlowHelpers/1.0.0/copyWithProgress.js`; the installer downloads it
|
||||
automatically.
|
||||
|
||||
@@ -222,7 +222,9 @@ const AUDIO_INPUTS = {
|
||||
ok(copyHelperCode.includes('maxRetries = 2'), 'copy helper retries failed writes');
|
||||
ok(copyHelperCode.includes('Copy size mismatch'), 'copy helper verifies destination size');
|
||||
ok(copyHelperCode.includes('fs.unlinkSync(destination)'), 'copy helper removes partial copies before retrying');
|
||||
ok(copyHelperCode.includes('COPY_BUFFER_SIZE = 4 * 1024 * 1024'), 'copy helper uses large SMB-friendly buffers');
|
||||
ok(copyHelperCode.includes('fs.promises.copyFile(source, destination)'), 'copy helper uses native file copying');
|
||||
ok(copyHelperCode.includes("copyMethod = 'auto'"), 'copy helper defaults to automatic copy selection');
|
||||
ok(copyHelperCode.includes("copyMethod === 'stream'"), 'copy helper supports stream fallback');
|
||||
|
||||
// ── Summary ───────────────────────────────────────────────────────────────
|
||||
console.log(`\n${'='.repeat(55)}`);
|
||||
|
||||
Reference in new issue
Block a user