OpenCloud
/
PeerTube
mirror of https://github.com/Chocobozzz/PeerTube
1
0
Fork 0
Code Issues Releases Wiki Activity

Fix concurrency issues when sendin chunks

pull/6544/head
Chocobozzz 2024-08-06 10:57:19 +02:00
parent 0ac25c0531
commit c5255e784c
No known key found for this signature in database
GPG Key ID: 583A612D890159BE
1 changed files with 30 additions and 20 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

16
apps/peertube-runner/src/server/process/shared/process-live.ts
View File

@ -268,14 +268,17 @@ export class ProcessLiveRTMPHLSTranscoding {
private async sendPendingChunks (): Promise<any> { private async sendPendingChunks (): Promise<any> {
if (this.ended) return Promise.resolve() if (this.ended) return Promise.resolve()
const promises: Promise<any>[] = [] const parallelPromises: Promise<any>[] = []
for (const playlist of this.pendingChunksPerPlaylist.keys()) { for (const playlist of this.pendingChunksPerPlaylist.keys()) {
let sequentialPromises: Promise<any>
for (const chunk of this.pendingChunksPerPlaylist.get(playlist)) { for (const chunk of this.pendingChunksPerPlaylist.get(playlist)) {
logger.debug(`Sending added live chunk ${chunk} update`) logger.debug(`Sending added live chunk ${chunk} update`)
const videoChunkFilename = basename(chunk) const videoChunkFilename = basename(chunk)
const payloadBuilder = async () => {
let payload: CustomLiveRTMPHLSTranscodingUpdatePayload = { let payload: CustomLiveRTMPHLSTranscodingUpdatePayload = {
type: 'add-chunk', type: 'add-chunk',
videoChunkFilename, videoChunkFilename,
@ -296,13 +299,20 @@ export class ProcessLiveRTMPHLSTranscoding {
} }
} }
promises.push(this.updateWithRetry(payload)) return payload
} }
const p = payloadBuilder().then(p => this.updateWithRetry(p))
if (!sequentialPromises) sequentialPromises = p
else sequentialPromises = sequentialPromises.then(() => p)
}
parallelPromises.push(sequentialPromises)
this.pendingChunksPerPlaylist.set(playlist, []) this.pendingChunksPerPlaylist.set(playlist, [])
} }
await Promise.all(promises) await Promise.all(parallelPromises)
} }
private async updateWithRetry (payload: CustomLiveRTMPHLSTranscodingUpdatePayload, currentTry = 1): Promise<any> { private async updateWithRetry (payload: CustomLiveRTMPHLSTranscodingUpdatePayload, currentTry = 1): Promise<any> {