X-Git-Url: https://git.immae.eu/?a=blobdiff_plain;f=server%2Flib%2Flive%2Fshared%2Fmuxing-session.ts;h=f3f8fc8863fc9c65ef1f8ed8c6f959159aef9223;hb=0c9668f77901e7540e2c7045eb0f2974a4842a69;hp=4c27d5dd885418299675e3645038a6783f2a9162;hpb=91a4893063402d7beabb3104f9b989b8f88b6038;p=github%2FChocobozzz%2FPeerTube.git diff --git a/server/lib/live/shared/muxing-session.ts b/server/lib/live/shared/muxing-session.ts index 4c27d5dd8..f3f8fc886 100644 --- a/server/lib/live/shared/muxing-session.ts +++ b/server/lib/live/shared/muxing-session.ts @@ -1,38 +1,42 @@ - import { mapSeries } from 'bluebird' import { FSWatcher, watch } from 'chokidar' -import { FfmpegCommand } from 'fluent-ffmpeg' +import { EventEmitter } from 'events' import { appendFile, ensureDir, readFile, stat } from 'fs-extra' +import PQueue from 'p-queue' import { basename, join } from 'path' -import { EventEmitter } from 'stream' -import { getLiveMuxingCommand, getLiveTranscodingCommand } from '@server/helpers/ffmpeg' +import { computeOutputFPS } from '@server/helpers/ffmpeg' import { logger, loggerTagsFactory, LoggerTagsFn } from '@server/helpers/logger' import { CONFIG } from '@server/initializers/config' -import { MEMOIZE_TTL, VIDEO_LIVE } from '@server/initializers/constants' -import { removeHLSFileObjectStorage, storeHLSFileFromFilename, storeHLSFileFromPath } from '@server/lib/object-storage' +import { MEMOIZE_TTL, P2P_MEDIA_LOADER_PEER_VERSION, VIDEO_LIVE } from '@server/initializers/constants' +import { removeHLSFileObjectStorageByPath, storeHLSFileFromFilename, storeHLSFileFromPath } from '@server/lib/object-storage' import { VideoFileModel } from '@server/models/video/video-file' +import { VideoStreamingPlaylistModel } from '@server/models/video/video-streaming-playlist' import { MStreamingPlaylistVideo, MUserId, MVideoLiveVideo } from '@server/types/models' -import { VideoStorage } from '@shared/models' -import { getLiveDirectory, getLiveReplayBaseDirectory } from '../../paths' -import { VideoTranscodingProfilesManager } from '../../transcoding/default-transcoding-profiles' +import { VideoStorage, VideoStreamingPlaylistType } from '@shared/models' +import { + generateHLSMasterPlaylistFilename, + generateHlsSha256SegmentsFilename, + getLiveDirectory, + getLiveReplayBaseDirectory +} from '../../paths' import { isAbleToUploadVideo } from '../../user' import { LiveQuotaStore } from '../live-quota-store' import { LiveSegmentShaStore } from '../live-segment-sha-store' -import { buildConcatenatedName } from '../live-utils' +import { buildConcatenatedName, getLiveSegmentTime } from '../live-utils' +import { AbstractTranscodingWrapper, FFmpegTranscodingWrapper, RemoteTranscodingWrapper } from './transcoding-wrapper' import memoizee = require('memoizee') - interface MuxingSessionEvents { - 'live-ready': (options: { videoId: number }) => void + 'live-ready': (options: { videoUUID: string }) => void - 'bad-socket-health': (options: { videoId: number }) => void - 'duration-exceeded': (options: { videoId: number }) => void - 'quota-exceeded': (options: { videoId: number }) => void + 'bad-socket-health': (options: { videoUUID: string }) => void + 'duration-exceeded': (options: { videoUUID: string }) => void + 'quota-exceeded': (options: { videoUUID: string }) => void - 'ffmpeg-end': (options: { videoId: number }) => void - 'ffmpeg-error': (options: { videoId: number }) => void + 'transcoding-end': (options: { videoUUID: string }) => void + 'transcoding-error': (options: { videoUUID: string }) => void - 'after-cleanup': (options: { videoId: number }) => void + 'after-cleanup': (options: { videoUUID: string }) => void } declare interface MuxingSession { @@ -47,13 +51,12 @@ declare interface MuxingSession { class MuxingSession extends EventEmitter { - private ffmpegCommand: FfmpegCommand + private transcodingWrapper: AbstractTranscodingWrapper private readonly context: any private readonly user: MUserId private readonly sessionId: string private readonly videoLive: MVideoLiveVideo - private readonly streamingPlaylist: MStreamingPlaylistVideo private readonly inputUrl: string private readonly fps: number private readonly allResolutions: number[] @@ -63,19 +66,19 @@ class MuxingSession extends EventEmitter { private readonly hasAudio: boolean - private readonly videoId: number private readonly videoUUID: string private readonly saveReplay: boolean private readonly outDirectory: string private readonly replayDirectory: string - private readonly liveSegmentShaStore: LiveSegmentShaStore - private readonly lTags: LoggerTagsFn private segmentsToProcessPerPlaylist: { [playlistId: string]: string[] } = {} + private streamingPlaylist: MStreamingPlaylistVideo + private liveSegmentShaStore: LiveSegmentShaStore + private tsWatcher: FSWatcher private masterWatcher: FSWatcher private m3u8Watcher: FSWatcher @@ -98,7 +101,6 @@ class MuxingSession extends EventEmitter { user: MUserId sessionId: string videoLive: MVideoLiveVideo - streamingPlaylist: MStreamingPlaylistVideo inputUrl: string fps: number bitrate: number @@ -112,7 +114,6 @@ class MuxingSession extends EventEmitter { this.user = options.user this.sessionId = options.sessionId this.videoLive = options.videoLive - this.streamingPlaylist = options.streamingPlaylist this.inputUrl = options.inputUrl this.fps = options.fps @@ -123,7 +124,6 @@ class MuxingSession extends EventEmitter { this.allResolutions = options.allResolutions - this.videoId = this.videoLive.Video.id this.videoUUID = this.videoLive.Video.uuid this.saveReplay = this.videoLive.saveReplay @@ -131,78 +131,34 @@ class MuxingSession extends EventEmitter { this.outDirectory = getLiveDirectory(this.videoLive.Video) this.replayDirectory = join(getLiveReplayBaseDirectory(this.videoLive.Video), new Date().toISOString()) - this.liveSegmentShaStore = new LiveSegmentShaStore({ - videoUUID: this.videoLive.Video.uuid, - sha256Path: join(this.outDirectory, this.streamingPlaylist.segmentsSha256Filename), - streamingPlaylist: this.streamingPlaylist, - sendToObjectStorage: CONFIG.OBJECT_STORAGE.ENABLED - }) - this.lTags = loggerTagsFactory('live', this.sessionId, this.videoUUID) } async runMuxing () { + this.streamingPlaylist = await this.createLivePlaylist() + + this.createLiveShaStore() this.createFiles() await this.prepareDirectories() - this.ffmpegCommand = CONFIG.LIVE.TRANSCODING.ENABLED - ? await getLiveTranscodingCommand({ - inputUrl: this.inputUrl, - - outPath: this.outDirectory, - masterPlaylistName: this.streamingPlaylist.playlistFilename, - - latencyMode: this.videoLive.latencyMode, - - resolutions: this.allResolutions, - fps: this.fps, - bitrate: this.bitrate, - ratio: this.ratio, - - hasAudio: this.hasAudio, + this.transcodingWrapper = this.buildTranscodingWrapper() - availableEncoders: VideoTranscodingProfilesManager.Instance.getAvailableEncoders(), - profile: CONFIG.LIVE.TRANSCODING.PROFILE - }) - : getLiveMuxingCommand({ - inputUrl: this.inputUrl, - outPath: this.outDirectory, - masterPlaylistName: this.streamingPlaylist.playlistFilename, - latencyMode: this.videoLive.latencyMode - }) + this.transcodingWrapper.on('end', () => this.onTranscodedEnded()) + this.transcodingWrapper.on('error', () => this.onTranscodingError()) - logger.info('Running live muxing/transcoding for %s.', this.videoUUID, this.lTags()) + await this.transcodingWrapper.run() this.watchMasterFile() this.watchTSFiles() this.watchM3U8File() - - let ffmpegShellCommand: string - this.ffmpegCommand.on('start', cmdline => { - ffmpegShellCommand = cmdline - - logger.debug('Running ffmpeg command for live', { ffmpegShellCommand, ...this.lTags() }) - }) - - this.ffmpegCommand.on('error', (err, stdout, stderr) => { - this.onFFmpegError({ err, stdout, stderr, ffmpegShellCommand }) - }) - - this.ffmpegCommand.on('end', () => { - this.emit('ffmpeg-end', ({ videoId: this.videoId })) - - this.onFFmpegEnded() - }) - - this.ffmpegCommand.run() } abort () { - if (!this.ffmpegCommand) return + if (!this.transcodingWrapper) return this.aborted = true - this.ffmpegCommand.kill('SIGINT') + this.transcodingWrapper.abort() } destroy () { @@ -211,65 +167,28 @@ class MuxingSession extends EventEmitter { this.hasClientSocketInBadHealthWithCache.clear() } - private onFFmpegError (options: { - err: any - stdout: string - stderr: string - ffmpegShellCommand: string - }) { - const { err, stdout, stderr, ffmpegShellCommand } = options - - this.onFFmpegEnded() - - // Don't care that we killed the ffmpeg process - if (err?.message?.includes('Exiting normally')) return - - logger.error('Live transcoding error.', { err, stdout, stderr, ffmpegShellCommand, ...this.lTags() }) - - this.emit('ffmpeg-error', ({ videoId: this.videoId })) - } - - private onFFmpegEnded () { - logger.info('RTMP transmuxing for video %s ended. Scheduling cleanup', this.inputUrl, this.lTags()) - - setTimeout(() => { - // Wait latest segments generation, and close watchers - - Promise.all([ this.tsWatcher.close(), this.masterWatcher.close(), this.m3u8Watcher.close() ]) - .then(() => { - // Process remaining segments hash - for (const key of Object.keys(this.segmentsToProcessPerPlaylist)) { - this.processSegments(this.segmentsToProcessPerPlaylist[key]) - } - }) - .catch(err => { - logger.error( - 'Cannot close watchers of %s or process remaining hash segments.', this.outDirectory, - { err, ...this.lTags() } - ) - }) - - this.emit('after-cleanup', { videoId: this.videoId }) - }, 1000) - } - private watchMasterFile () { this.masterWatcher = watch(this.outDirectory + '/' + this.streamingPlaylist.playlistFilename) this.masterWatcher.on('add', async () => { - if (this.streamingPlaylist.storage === VideoStorage.OBJECT_STORAGE) { - try { + try { + if (this.streamingPlaylist.storage === VideoStorage.OBJECT_STORAGE) { const url = await storeHLSFileFromFilename(this.streamingPlaylist, this.streamingPlaylist.playlistFilename) this.streamingPlaylist.playlistUrl = url - await this.streamingPlaylist.save() - } catch (err) { - logger.error('Cannot upload live master file to object storage.', { err, ...this.lTags() }) } + + this.streamingPlaylist.assignP2PMediaLoaderInfoHashes(this.videoLive.Video, this.allResolutions) + + await this.streamingPlaylist.save() + } catch (err) { + logger.error('Cannot update streaming playlist.', { err, ...this.lTags() }) } this.masterPlaylistCreated = true + logger.info('Master playlist file for %s has been created', this.videoUUID, this.lTags()) + this.masterWatcher.close() .catch(err => logger.error('Cannot close master watcher of %s.', this.outDirectory, { err, ...this.lTags() })) }) @@ -278,11 +197,18 @@ class MuxingSession extends EventEmitter { private watchM3U8File () { this.m3u8Watcher = watch(this.outDirectory + '/*.m3u8') + const sendQueues = new Map() + const onChangeOrAdd = async (m3u8Path: string) => { if (this.streamingPlaylist.storage !== VideoStorage.OBJECT_STORAGE) return try { - await storeHLSFileFromPath(this.streamingPlaylist, m3u8Path) + if (!sendQueues.has(m3u8Path)) { + sendQueues.set(m3u8Path, new PQueue({ concurrency: 1 })) + } + + const queue = sendQueues.get(m3u8Path) + await queue.add(() => storeHLSFileFromPath(this.streamingPlaylist, m3u8Path)) } catch (err) { logger.error('Cannot store in object storage m3u8 file %s', m3u8Path, { err, ...this.lTags() }) } @@ -309,19 +235,19 @@ class MuxingSession extends EventEmitter { this.segmentsToProcessPerPlaylist[playlistId] = [ segmentPath ] if (this.hasClientSocketInBadHealthWithCache(this.sessionId)) { - this.emit('bad-socket-health', { videoId: this.videoId }) + this.emit('bad-socket-health', { videoUUID: this.videoUUID }) return } // Duration constraint check if (this.isDurationConstraintValid(startStreamDateTime) !== true) { - this.emit('duration-exceeded', { videoId: this.videoId }) + this.emit('duration-exceeded', { videoUUID: this.videoUUID }) return } // Check user quota if the user enabled replay saving if (await this.isQuotaExceeded(segmentPath) === true) { - this.emit('quota-exceeded', { videoId: this.videoId }) + this.emit('quota-exceeded', { videoUUID: this.videoUUID }) } } @@ -334,7 +260,7 @@ class MuxingSession extends EventEmitter { if (this.streamingPlaylist.storage === VideoStorage.OBJECT_STORAGE) { try { - await removeHLSFileObjectStorage(this.streamingPlaylist, segmentPath) + await removeHLSFileObjectStorageByPath(this.streamingPlaylist, segmentPath) } catch (err) { logger.error('Cannot remove segment %s from object storage', segmentPath, { err, ...this.lTags() }) } @@ -429,10 +355,40 @@ class MuxingSession extends EventEmitter { if (this.masterPlaylistCreated && !this.liveReady) { this.liveReady = true - this.emit('live-ready', { videoId: this.videoId }) + this.emit('live-ready', { videoUUID: this.videoUUID }) } } + private onTranscodingError () { + this.emit('transcoding-error', ({ videoUUID: this.videoUUID })) + } + + private onTranscodedEnded () { + this.emit('transcoding-end', ({ videoUUID: this.videoUUID })) + + logger.info('RTMP transmuxing for video %s ended. Scheduling cleanup', this.inputUrl, this.lTags()) + + setTimeout(() => { + // Wait latest segments generation, and close watchers + + Promise.all([ this.tsWatcher.close(), this.masterWatcher.close(), this.m3u8Watcher.close() ]) + .then(() => { + // Process remaining segments hash + for (const key of Object.keys(this.segmentsToProcessPerPlaylist)) { + this.processSegments(this.segmentsToProcessPerPlaylist[key]) + } + }) + .catch(err => { + logger.error( + 'Cannot close watchers of %s or process remaining hash segments.', this.outDirectory, + { err, ...this.lTags() } + ) + }) + + this.emit('after-cleanup', { videoUUID: this.videoUUID }) + }, 1000) + } + private hasClientSocketInBadHealth (sessionId: string) { const rtmpSession = this.context.sessions.get(sessionId) @@ -469,6 +425,61 @@ class MuxingSession extends EventEmitter { logger.error('Cannot copy segment %s to replay directory.', segmentPath, { err, ...this.lTags() }) } } + + private async createLivePlaylist (): Promise { + const playlist = await VideoStreamingPlaylistModel.loadOrGenerate(this.videoLive.Video) + + playlist.playlistFilename = generateHLSMasterPlaylistFilename(true) + playlist.segmentsSha256Filename = generateHlsSha256SegmentsFilename(true) + + playlist.p2pMediaLoaderPeerVersion = P2P_MEDIA_LOADER_PEER_VERSION + playlist.type = VideoStreamingPlaylistType.HLS + + playlist.storage = CONFIG.OBJECT_STORAGE.ENABLED + ? VideoStorage.OBJECT_STORAGE + : VideoStorage.FILE_SYSTEM + + return playlist.save() + } + + private createLiveShaStore () { + this.liveSegmentShaStore = new LiveSegmentShaStore({ + videoUUID: this.videoLive.Video.uuid, + sha256Path: join(this.outDirectory, this.streamingPlaylist.segmentsSha256Filename), + streamingPlaylist: this.streamingPlaylist, + sendToObjectStorage: CONFIG.OBJECT_STORAGE.ENABLED + }) + } + + private buildTranscodingWrapper () { + const options = { + streamingPlaylist: this.streamingPlaylist, + videoLive: this.videoLive, + + lTags: this.lTags, + + inputUrl: this.inputUrl, + + toTranscode: this.allResolutions.map(resolution => ({ + resolution, + fps: computeOutputFPS({ inputFPS: this.fps, resolution }) + })), + + fps: this.fps, + bitrate: this.bitrate, + ratio: this.ratio, + hasAudio: this.hasAudio, + + segmentListSize: VIDEO_LIVE.SEGMENTS_LIST_SIZE, + segmentDuration: getLiveSegmentTime(this.videoLive.latencyMode), + + outDirectory: this.outDirectory + } + + return CONFIG.LIVE.TRANSCODING.ENABLED && CONFIG.LIVE.TRANSCODING.REMOTE_RUNNERS.ENABLED + ? new RemoteTranscodingWrapper(options) + : new FFmpegTranscodingWrapper(options) + } } // ---------------------------------------------------------------------------