Add priority to transcoding jobs
(1 = highest priority) 100 for new resolutions 10 for original file optimization Add a malus for transcoding jobs depending on how many uploads the user did in the last 7 days
This commit is contained in:
parent
92c871b405
commit
77d7e851dc
15 changed files with 172 additions and 79 deletions
|
@ -40,6 +40,7 @@
|
||||||
<th style="width: 40px"></th>
|
<th style="width: 40px"></th>
|
||||||
<th style="width: calc(100% - 390px)" class="job-id" i18n>ID</th>
|
<th style="width: calc(100% - 390px)" class="job-id" i18n>ID</th>
|
||||||
<th style="width: 200px" class="job-type" i18n>Type</th>
|
<th style="width: 200px" class="job-type" i18n>Type</th>
|
||||||
|
<th style="width: 200px" class="job-priority" i18n>Priority <small>(1 = highest priority)</small></th>
|
||||||
<th style="width: 200px" class="job-state" i18n *ngIf="jobState === 'all'">State</th>
|
<th style="width: 200px" class="job-state" i18n *ngIf="jobState === 'all'">State</th>
|
||||||
<th style="width: 100px" class="job-progress" i18n *ngIf="hasProgress()">Progress</th>
|
<th style="width: 100px" class="job-progress" i18n *ngIf="hasProgress()">Progress</th>
|
||||||
<th style="width: 150px" class="job-date" i18n pSortableColumn="createdAt">Created <p-sortIcon field="createdAt"></p-sortIcon></th>
|
<th style="width: 150px" class="job-date" i18n pSortableColumn="createdAt">Created <p-sortIcon field="createdAt"></p-sortIcon></th>
|
||||||
|
@ -56,12 +57,13 @@
|
||||||
|
|
||||||
<td class="job-id c-hand" [pRowToggler]="job" [title]="job.id">{{ job.id }}</td>
|
<td class="job-id c-hand" [pRowToggler]="job" [title]="job.id">{{ job.id }}</td>
|
||||||
<td class="job-type c-hand" [pRowToggler]="job">{{ job.type }}</td>
|
<td class="job-type c-hand" [pRowToggler]="job">{{ job.type }}</td>
|
||||||
|
<td class="job-priority c-hand" [pRowToggler]="job">{{ job.priority }}</td>
|
||||||
|
|
||||||
<td class="job-state c-hand" [pRowToggler]="job" *ngIf="jobState === 'all'">
|
<td class="job-state c-hand" [pRowToggler]="job" *ngIf="jobState === 'all'">
|
||||||
<span class="badge" [ngClass]="getJobStateClass(job.state)">{{ job.state }}</span>
|
<span class="badge" [ngClass]="getJobStateClass(job.state)">{{ job.state }}</span>
|
||||||
</td>
|
</td>
|
||||||
|
|
||||||
<td class="job-state" [pRowToggler]="job" *ngIf="hasProgress()">
|
<td class="job-progress c-hand" [pRowToggler]="job" *ngIf="hasProgress()">
|
||||||
{{ getProgress(job) }}
|
{{ getProgress(job) }}
|
||||||
</td>
|
</td>
|
||||||
|
|
||||||
|
|
|
@ -10,7 +10,8 @@
|
||||||
}
|
}
|
||||||
|
|
||||||
.job-type,
|
.job-type,
|
||||||
.job-state {
|
.job-state,
|
||||||
|
.job-priority {
|
||||||
width: 150px !important;
|
width: 150px !important;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
|
@ -21,19 +21,19 @@ export class JobsComponent extends RestTable implements OnInit {
|
||||||
|
|
||||||
jobType: JobTypeClient = 'all'
|
jobType: JobTypeClient = 'all'
|
||||||
jobTypes: JobTypeClient[] = [
|
jobTypes: JobTypeClient[] = [
|
||||||
'all',
|
|
||||||
'activitypub-follow',
|
'activitypub-follow',
|
||||||
'activitypub-http-broadcast',
|
'activitypub-http-broadcast',
|
||||||
'activitypub-http-fetcher',
|
'activitypub-http-fetcher',
|
||||||
'activitypub-http-unicast',
|
'activitypub-http-unicast',
|
||||||
|
'activitypub-refresher',
|
||||||
|
'all',
|
||||||
'email',
|
'email',
|
||||||
'video-transcoding',
|
|
||||||
'video-file-import',
|
'video-file-import',
|
||||||
'video-import',
|
'video-import',
|
||||||
'videos-views',
|
'video-live-ending',
|
||||||
'activitypub-refresher',
|
|
||||||
'video-redundancy',
|
'video-redundancy',
|
||||||
'video-live-ending'
|
'video-transcoding',
|
||||||
|
'videos-views'
|
||||||
]
|
]
|
||||||
|
|
||||||
jobs: Job[] = []
|
jobs: Job[] = []
|
||||||
|
@ -75,7 +75,7 @@ export class JobsComponent extends RestTable implements OnInit {
|
||||||
getColspan () {
|
getColspan () {
|
||||||
if (this.jobState === 'all' && this.hasProgress()) return 6
|
if (this.jobState === 'all' && this.hasProgress()) return 6
|
||||||
|
|
||||||
if (this.jobState === 'all') return 5
|
if (this.jobState === 'all' || this.hasProgress()) return 5
|
||||||
|
|
||||||
return 4
|
return 4
|
||||||
}
|
}
|
||||||
|
|
|
@ -69,6 +69,7 @@ async function formatJob (job: any, state?: JobState): Promise<Job> {
|
||||||
type: job.queue.name as JobType,
|
type: job.queue.name as JobType,
|
||||||
data: job.data,
|
data: job.data,
|
||||||
progress: await job.progress(),
|
progress: await job.progress(),
|
||||||
|
priority: job.opts.priority,
|
||||||
error,
|
error,
|
||||||
createdAt: new Date(job.timestamp),
|
createdAt: new Date(job.timestamp),
|
||||||
finishedOn: new Date(job.finishedOn),
|
finishedOn: new Date(job.finishedOn),
|
||||||
|
|
|
@ -2,16 +2,16 @@ import * as express from 'express'
|
||||||
import { move } from 'fs-extra'
|
import { move } from 'fs-extra'
|
||||||
import { extname } from 'path'
|
import { extname } from 'path'
|
||||||
import toInt from 'validator/lib/toInt'
|
import toInt from 'validator/lib/toInt'
|
||||||
import { addOptimizeOrMergeAudioJob } from '@server/helpers/video'
|
|
||||||
import { createTorrentAndSetInfoHash } from '@server/helpers/webtorrent'
|
import { createTorrentAndSetInfoHash } from '@server/helpers/webtorrent'
|
||||||
import { changeVideoChannelShare } from '@server/lib/activitypub/share'
|
import { changeVideoChannelShare } from '@server/lib/activitypub/share'
|
||||||
import { getLocalVideoActivityPubUrl } from '@server/lib/activitypub/url'
|
import { getLocalVideoActivityPubUrl } from '@server/lib/activitypub/url'
|
||||||
import { LiveManager } from '@server/lib/live-manager'
|
import { LiveManager } from '@server/lib/live-manager'
|
||||||
import { buildLocalVideoFromReq, buildVideoThumbnailsFromReq, setVideoTags } from '@server/lib/video'
|
import { addOptimizeOrMergeAudioJob, buildLocalVideoFromReq, buildVideoThumbnailsFromReq, setVideoTags } from '@server/lib/video'
|
||||||
import { getVideoFilePath } from '@server/lib/video-paths'
|
import { getVideoFilePath } from '@server/lib/video-paths'
|
||||||
import { getServerActor } from '@server/models/application/application'
|
import { getServerActor } from '@server/models/application/application'
|
||||||
import { MVideoFullLight } from '@server/types/models'
|
import { MVideoFullLight } from '@server/types/models'
|
||||||
import { VideoCreate, VideoState, VideoUpdate } from '../../../../shared'
|
import { VideoCreate, VideoState, VideoUpdate } from '../../../../shared'
|
||||||
|
import { HttpStatusCode } from '../../../../shared/core-utils/miscs/http-error-codes'
|
||||||
import { VideoFilter } from '../../../../shared/models/videos/video-query.type'
|
import { VideoFilter } from '../../../../shared/models/videos/video-query.type'
|
||||||
import { auditLoggerFactory, getAuditIdFromRes, VideoAuditView } from '../../../helpers/audit-logger'
|
import { auditLoggerFactory, getAuditIdFromRes, VideoAuditView } from '../../../helpers/audit-logger'
|
||||||
import { resetSequelizeInstance } from '../../../helpers/database-utils'
|
import { resetSequelizeInstance } from '../../../helpers/database-utils'
|
||||||
|
@ -66,7 +66,6 @@ import { liveRouter } from './live'
|
||||||
import { ownershipVideoRouter } from './ownership'
|
import { ownershipVideoRouter } from './ownership'
|
||||||
import { rateVideoRouter } from './rate'
|
import { rateVideoRouter } from './rate'
|
||||||
import { watchingRouter } from './watching'
|
import { watchingRouter } from './watching'
|
||||||
import { HttpStatusCode } from '../../../../shared/core-utils/miscs/http-error-codes'
|
|
||||||
|
|
||||||
const auditLogger = auditLoggerFactory('videos')
|
const auditLogger = auditLoggerFactory('videos')
|
||||||
const videosRouter = express.Router()
|
const videosRouter = express.Router()
|
||||||
|
@ -267,7 +266,7 @@ async function addVideo (req: express.Request, res: express.Response) {
|
||||||
Notifier.Instance.notifyOnNewVideoIfNeeded(videoCreated)
|
Notifier.Instance.notifyOnNewVideoIfNeeded(videoCreated)
|
||||||
|
|
||||||
if (video.state === VideoState.TO_TRANSCODE) {
|
if (video.state === VideoState.TO_TRANSCODE) {
|
||||||
await addOptimizeOrMergeAudioJob(videoCreated, videoFile)
|
await addOptimizeOrMergeAudioJob(videoCreated, videoFile, res.locals.oauth.token.User)
|
||||||
}
|
}
|
||||||
|
|
||||||
Hooks.runAction('action:api.video.uploaded', { video: videoCreated })
|
Hooks.runAction('action:api.video.uploaded', { video: videoCreated })
|
||||||
|
|
|
@ -4,6 +4,14 @@ import { Model } from 'sequelize-typescript'
|
||||||
import { logger } from './logger'
|
import { logger } from './logger'
|
||||||
import { Transaction } from 'sequelize'
|
import { Transaction } from 'sequelize'
|
||||||
|
|
||||||
|
function retryTransactionWrapper <T, A, B, C, D> (
|
||||||
|
functionToRetry: (arg1: A, arg2: B, arg3: C, arg4: D) => Promise<T> | Bluebird<T>,
|
||||||
|
arg1: A,
|
||||||
|
arg2: B,
|
||||||
|
arg3: C,
|
||||||
|
arg4: D,
|
||||||
|
): Promise<T>
|
||||||
|
|
||||||
function retryTransactionWrapper <T, A, B, C> (
|
function retryTransactionWrapper <T, A, B, C> (
|
||||||
functionToRetry: (arg1: A, arg2: B, arg3: C) => Promise<T> | Bluebird<T>,
|
functionToRetry: (arg1: A, arg2: B, arg3: C) => Promise<T> | Bluebird<T>,
|
||||||
arg1: A,
|
arg1: A,
|
||||||
|
|
|
@ -1,20 +1,17 @@
|
||||||
import { Response } from 'express'
|
import { Response } from 'express'
|
||||||
import { CONFIG } from '@server/initializers/config'
|
import { CONFIG } from '@server/initializers/config'
|
||||||
import { DEFAULT_AUDIO_RESOLUTION } from '@server/initializers/constants'
|
|
||||||
import { JobQueue } from '@server/lib/job-queue'
|
|
||||||
import {
|
import {
|
||||||
isStreamingPlaylist,
|
isStreamingPlaylist,
|
||||||
MStreamingPlaylistVideo,
|
MStreamingPlaylistVideo,
|
||||||
MVideo,
|
MVideo,
|
||||||
MVideoAccountLightBlacklistAllFiles,
|
MVideoAccountLightBlacklistAllFiles,
|
||||||
MVideoFile,
|
|
||||||
MVideoFullLight,
|
MVideoFullLight,
|
||||||
MVideoIdThumbnail,
|
MVideoIdThumbnail,
|
||||||
MVideoImmutable,
|
MVideoImmutable,
|
||||||
MVideoThumbnail,
|
MVideoThumbnail,
|
||||||
MVideoWithRights
|
MVideoWithRights
|
||||||
} from '@server/types/models'
|
} from '@server/types/models'
|
||||||
import { VideoPrivacy, VideoState, VideoTranscodingPayload } from '@shared/models'
|
import { VideoPrivacy, VideoState } from '@shared/models'
|
||||||
import { VideoModel } from '../models/video/video'
|
import { VideoModel } from '../models/video/video'
|
||||||
|
|
||||||
type VideoFetchType = 'all' | 'only-video' | 'only-video-with-rights' | 'id' | 'none' | 'only-immutable-attributes'
|
type VideoFetchType = 'all' | 'only-video' | 'only-video-with-rights' | 'id' | 'none' | 'only-immutable-attributes'
|
||||||
|
@ -69,27 +66,6 @@ function getVideoWithAttributes (res: Response) {
|
||||||
return res.locals.videoAll || res.locals.onlyVideo || res.locals.onlyVideoWithRights
|
return res.locals.videoAll || res.locals.onlyVideo || res.locals.onlyVideoWithRights
|
||||||
}
|
}
|
||||||
|
|
||||||
function addOptimizeOrMergeAudioJob (video: MVideo, videoFile: MVideoFile) {
|
|
||||||
let dataInput: VideoTranscodingPayload
|
|
||||||
|
|
||||||
if (videoFile.isAudio()) {
|
|
||||||
dataInput = {
|
|
||||||
type: 'merge-audio-to-webtorrent',
|
|
||||||
resolution: DEFAULT_AUDIO_RESOLUTION,
|
|
||||||
videoUUID: video.uuid,
|
|
||||||
isNewVideo: true
|
|
||||||
}
|
|
||||||
} else {
|
|
||||||
dataInput = {
|
|
||||||
type: 'optimize-to-webtorrent',
|
|
||||||
videoUUID: video.uuid,
|
|
||||||
isNewVideo: true
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
return JobQueue.Instance.createJobWithPromise({ type: 'video-transcoding', payload: dataInput })
|
|
||||||
}
|
|
||||||
|
|
||||||
function extractVideo (videoOrPlaylist: MVideo | MStreamingPlaylistVideo) {
|
function extractVideo (videoOrPlaylist: MVideo | MStreamingPlaylistVideo) {
|
||||||
return isStreamingPlaylist(videoOrPlaylist)
|
return isStreamingPlaylist(videoOrPlaylist)
|
||||||
? videoOrPlaylist.Video
|
? videoOrPlaylist.Video
|
||||||
|
@ -107,7 +83,6 @@ function isStateForFederation (state: VideoState) {
|
||||||
const castedState = parseInt(state + '', 10)
|
const castedState = parseInt(state + '', 10)
|
||||||
|
|
||||||
return castedState === VideoState.PUBLISHED || castedState === VideoState.WAITING_FOR_LIVE || castedState === VideoState.LIVE_ENDED
|
return castedState === VideoState.PUBLISHED || castedState === VideoState.WAITING_FOR_LIVE || castedState === VideoState.LIVE_ENDED
|
||||||
|
|
||||||
}
|
}
|
||||||
|
|
||||||
function getPrivaciesForFederation () {
|
function getPrivaciesForFederation () {
|
||||||
|
@ -130,7 +105,6 @@ export {
|
||||||
fetchVideo,
|
fetchVideo,
|
||||||
getVideoWithAttributes,
|
getVideoWithAttributes,
|
||||||
fetchVideoByUrl,
|
fetchVideoByUrl,
|
||||||
addOptimizeOrMergeAudioJob,
|
|
||||||
extractVideo,
|
extractVideo,
|
||||||
getExtFromMimetype,
|
getExtFromMimetype,
|
||||||
isStateForFederation,
|
isStateForFederation,
|
||||||
|
|
|
@ -179,6 +179,12 @@ const REPEAT_JOBS: { [ id: string ]: EveryRepeatOptions | CronRepeatOptions } =
|
||||||
cron: randomInt(1, 20) + ' * * * *' // Between 1-20 minutes past the hour
|
cron: randomInt(1, 20) + ' * * * *' // Between 1-20 minutes past the hour
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
const JOB_PRIORITY = {
|
||||||
|
TRANSCODING: {
|
||||||
|
OPTIMIZER: 10,
|
||||||
|
NEW_RESOLUTION: 100
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
const BROADCAST_CONCURRENCY = 10 // How many requests in parallel we do in activitypub-http-broadcast job
|
const BROADCAST_CONCURRENCY = 10 // How many requests in parallel we do in activitypub-http-broadcast job
|
||||||
const CRAWL_REQUEST_CONCURRENCY = 1 // How many requests in parallel to fetch remote data (likes, shares...)
|
const CRAWL_REQUEST_CONCURRENCY = 1 // How many requests in parallel to fetch remote data (likes, shares...)
|
||||||
|
@ -851,6 +857,7 @@ export {
|
||||||
VIDEO_STATES,
|
VIDEO_STATES,
|
||||||
QUEUE_CONCURRENCY,
|
QUEUE_CONCURRENCY,
|
||||||
VIDEO_RATE_TYPES,
|
VIDEO_RATE_TYPES,
|
||||||
|
JOB_PRIORITY,
|
||||||
VIDEO_TRANSCODING_FPS,
|
VIDEO_TRANSCODING_FPS,
|
||||||
FFMPEG_NICE,
|
FFMPEG_NICE,
|
||||||
ABUSE_STATES,
|
ABUSE_STATES,
|
||||||
|
|
|
@ -3,6 +3,7 @@ import { copy, stat } from 'fs-extra'
|
||||||
import { extname } from 'path'
|
import { extname } from 'path'
|
||||||
import { createTorrentAndSetInfoHash } from '@server/helpers/webtorrent'
|
import { createTorrentAndSetInfoHash } from '@server/helpers/webtorrent'
|
||||||
import { getVideoFilePath } from '@server/lib/video-paths'
|
import { getVideoFilePath } from '@server/lib/video-paths'
|
||||||
|
import { UserModel } from '@server/models/account/user'
|
||||||
import { MVideoFile, MVideoWithFile } from '@server/types/models'
|
import { MVideoFile, MVideoWithFile } from '@server/types/models'
|
||||||
import { VideoFileImportPayload } from '@shared/models'
|
import { VideoFileImportPayload } from '@shared/models'
|
||||||
import { getVideoFileFPS, getVideoFileResolution } from '../../../helpers/ffprobe-utils'
|
import { getVideoFileFPS, getVideoFileResolution } from '../../../helpers/ffprobe-utils'
|
||||||
|
@ -24,7 +25,9 @@ async function processVideoFileImport (job: Bull.Job) {
|
||||||
|
|
||||||
await updateVideoFile(video, payload.filePath)
|
await updateVideoFile(video, payload.filePath)
|
||||||
|
|
||||||
await onNewWebTorrentFileResolution(video)
|
const user = await UserModel.loadByChannelActorId(video.VideoChannel.actorId)
|
||||||
|
await onNewWebTorrentFileResolution(video, user)
|
||||||
|
|
||||||
return video
|
return video
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
|
@ -1,10 +1,10 @@
|
||||||
import * as Bull from 'bull'
|
import * as Bull from 'bull'
|
||||||
import { move, remove, stat } from 'fs-extra'
|
import { move, remove, stat } from 'fs-extra'
|
||||||
import { extname } from 'path'
|
import { extname } from 'path'
|
||||||
import { addOptimizeOrMergeAudioJob } from '@server/helpers/video'
|
|
||||||
import { isPostImportVideoAccepted } from '@server/lib/moderation'
|
import { isPostImportVideoAccepted } from '@server/lib/moderation'
|
||||||
import { Hooks } from '@server/lib/plugins/hooks'
|
import { Hooks } from '@server/lib/plugins/hooks'
|
||||||
import { isAbleToUploadVideo } from '@server/lib/user'
|
import { isAbleToUploadVideo } from '@server/lib/user'
|
||||||
|
import { addOptimizeOrMergeAudioJob } from '@server/lib/video'
|
||||||
import { getVideoFilePath } from '@server/lib/video-paths'
|
import { getVideoFilePath } from '@server/lib/video-paths'
|
||||||
import { MVideoImportDefault, MVideoImportDefaultFiles, MVideoImportVideo } from '@server/types/models/video/video-import'
|
import { MVideoImportDefault, MVideoImportDefaultFiles, MVideoImportVideo } from '@server/types/models/video/video-import'
|
||||||
import {
|
import {
|
||||||
|
@ -224,7 +224,7 @@ async function processFile (downloader: () => Promise<string>, videoImport: MVid
|
||||||
|
|
||||||
// Create transcoding jobs?
|
// Create transcoding jobs?
|
||||||
if (video.state === VideoState.TO_TRANSCODE) {
|
if (video.state === VideoState.TO_TRANSCODE) {
|
||||||
await addOptimizeOrMergeAudioJob(videoImportUpdated.Video, videoFile)
|
await addOptimizeOrMergeAudioJob(videoImportUpdated.Video, videoFile, videoImport.User)
|
||||||
}
|
}
|
||||||
|
|
||||||
} catch (err) {
|
} catch (err) {
|
||||||
|
|
|
@ -1,8 +1,9 @@
|
||||||
import * as Bull from 'bull'
|
import * as Bull from 'bull'
|
||||||
import { TranscodeOptionsType } from '@server/helpers/ffmpeg-utils'
|
import { TranscodeOptionsType } from '@server/helpers/ffmpeg-utils'
|
||||||
import { publishAndFederateIfNeeded } from '@server/lib/video'
|
import { JOB_PRIORITY } from '@server/initializers/constants'
|
||||||
|
import { getJobTranscodingPriorityMalus, publishAndFederateIfNeeded } from '@server/lib/video'
|
||||||
import { getVideoFilePath } from '@server/lib/video-paths'
|
import { getVideoFilePath } from '@server/lib/video-paths'
|
||||||
import { MVideoFullLight, MVideoUUID, MVideoWithFile } from '@server/types/models'
|
import { MUser, MUserId, MVideoFullLight, MVideoUUID, MVideoWithFile } from '@server/types/models'
|
||||||
import {
|
import {
|
||||||
HLSTranscodingPayload,
|
HLSTranscodingPayload,
|
||||||
MergeAudioTranscodingPayload,
|
MergeAudioTranscodingPayload,
|
||||||
|
@ -25,8 +26,11 @@ import {
|
||||||
transcodeNewWebTorrentResolution
|
transcodeNewWebTorrentResolution
|
||||||
} from '../../video-transcoding'
|
} from '../../video-transcoding'
|
||||||
import { JobQueue } from '../job-queue'
|
import { JobQueue } from '../job-queue'
|
||||||
|
import { UserModel } from '@server/models/account/user'
|
||||||
|
|
||||||
const handlers: { [ id: string ]: (job: Bull.Job, payload: VideoTranscodingPayload, video: MVideoFullLight) => Promise<any> } = {
|
type HandlerFunction = (job: Bull.Job, payload: VideoTranscodingPayload, video: MVideoFullLight, user: MUser) => Promise<any>
|
||||||
|
|
||||||
|
const handlers: { [ id: string ]: HandlerFunction } = {
|
||||||
// Deprecated, introduced in 3.1
|
// Deprecated, introduced in 3.1
|
||||||
'hls': handleHLSJob,
|
'hls': handleHLSJob,
|
||||||
'new-resolution-to-hls': handleHLSJob,
|
'new-resolution-to-hls': handleHLSJob,
|
||||||
|
@ -55,13 +59,15 @@ async function processVideoTranscoding (job: Bull.Job) {
|
||||||
return undefined
|
return undefined
|
||||||
}
|
}
|
||||||
|
|
||||||
|
const user = await UserModel.loadByChannelActorId(video.VideoChannel.actorId)
|
||||||
|
|
||||||
const handler = handlers[payload.type]
|
const handler = handlers[payload.type]
|
||||||
|
|
||||||
if (!handler) {
|
if (!handler) {
|
||||||
throw new Error('Cannot find transcoding handler for ' + payload.type)
|
throw new Error('Cannot find transcoding handler for ' + payload.type)
|
||||||
}
|
}
|
||||||
|
|
||||||
await handler(job, payload, video)
|
await handler(job, payload, video, user)
|
||||||
|
|
||||||
return video
|
return video
|
||||||
}
|
}
|
||||||
|
@ -90,22 +96,27 @@ async function handleHLSJob (job: Bull.Job, payload: HLSTranscodingPayload, vide
|
||||||
await retryTransactionWrapper(onHlsPlaylistGeneration, video)
|
await retryTransactionWrapper(onHlsPlaylistGeneration, video)
|
||||||
}
|
}
|
||||||
|
|
||||||
async function handleNewWebTorrentResolutionJob (job: Bull.Job, payload: NewResolutionTranscodingPayload, video: MVideoFullLight) {
|
async function handleNewWebTorrentResolutionJob (
|
||||||
|
job: Bull.Job,
|
||||||
|
payload: NewResolutionTranscodingPayload,
|
||||||
|
video: MVideoFullLight,
|
||||||
|
user: MUserId
|
||||||
|
) {
|
||||||
await transcodeNewWebTorrentResolution(video, payload.resolution, payload.isPortraitMode || false, job)
|
await transcodeNewWebTorrentResolution(video, payload.resolution, payload.isPortraitMode || false, job)
|
||||||
|
|
||||||
await retryTransactionWrapper(onNewWebTorrentFileResolution, video, payload)
|
await retryTransactionWrapper(onNewWebTorrentFileResolution, video, user, payload)
|
||||||
}
|
}
|
||||||
|
|
||||||
async function handleWebTorrentMergeAudioJob (job: Bull.Job, payload: MergeAudioTranscodingPayload, video: MVideoFullLight) {
|
async function handleWebTorrentMergeAudioJob (job: Bull.Job, payload: MergeAudioTranscodingPayload, video: MVideoFullLight, user: MUserId) {
|
||||||
await mergeAudioVideofile(video, payload.resolution, job)
|
await mergeAudioVideofile(video, payload.resolution, job)
|
||||||
|
|
||||||
await retryTransactionWrapper(onNewWebTorrentFileResolution, video, payload)
|
await retryTransactionWrapper(onNewWebTorrentFileResolution, video, user, payload)
|
||||||
}
|
}
|
||||||
|
|
||||||
async function handleWebTorrentOptimizeJob (job: Bull.Job, payload: OptimizeTranscodingPayload, video: MVideoFullLight) {
|
async function handleWebTorrentOptimizeJob (job: Bull.Job, payload: OptimizeTranscodingPayload, video: MVideoFullLight, user: MUserId) {
|
||||||
const transcodeType = await optimizeOriginalVideofile(video, video.getMaxQualityFile(), job)
|
const transcodeType = await optimizeOriginalVideofile(video, video.getMaxQualityFile(), job)
|
||||||
|
|
||||||
await retryTransactionWrapper(onVideoFileOptimizer, video, payload, transcodeType)
|
await retryTransactionWrapper(onVideoFileOptimizer, video, payload, transcodeType, user)
|
||||||
}
|
}
|
||||||
|
|
||||||
// ---------------------------------------------------------------------------
|
// ---------------------------------------------------------------------------
|
||||||
|
@ -129,7 +140,8 @@ async function onHlsPlaylistGeneration (video: MVideoFullLight) {
|
||||||
async function onVideoFileOptimizer (
|
async function onVideoFileOptimizer (
|
||||||
videoArg: MVideoWithFile,
|
videoArg: MVideoWithFile,
|
||||||
payload: OptimizeTranscodingPayload,
|
payload: OptimizeTranscodingPayload,
|
||||||
transcodeType: TranscodeOptionsType
|
transcodeType: TranscodeOptionsType,
|
||||||
|
user: MUserId
|
||||||
) {
|
) {
|
||||||
if (videoArg === undefined) return undefined
|
if (videoArg === undefined) return undefined
|
||||||
|
|
||||||
|
@ -142,13 +154,6 @@ async function onVideoFileOptimizer (
|
||||||
// Video does not exist anymore
|
// Video does not exist anymore
|
||||||
if (!videoDatabase) return undefined
|
if (!videoDatabase) return undefined
|
||||||
|
|
||||||
// Create transcoding jobs if there are enabled resolutions
|
|
||||||
const resolutionsEnabled = computeResolutionsToTranscode(videoFileResolution, 'vod')
|
|
||||||
logger.info(
|
|
||||||
'Resolutions computed for video %s and origin file resolution of %d.', videoDatabase.uuid, videoFileResolution,
|
|
||||||
{ resolutions: resolutionsEnabled }
|
|
||||||
)
|
|
||||||
|
|
||||||
let videoPublished = false
|
let videoPublished = false
|
||||||
|
|
||||||
// Generate HLS version of the original file
|
// Generate HLS version of the original file
|
||||||
|
@ -158,9 +163,9 @@ async function onVideoFileOptimizer (
|
||||||
// If we quick transcoded original file, force transcoding for HLS to avoid some weird playback issues
|
// If we quick transcoded original file, force transcoding for HLS to avoid some weird playback issues
|
||||||
copyCodecs: transcodeType !== 'quick-transcode'
|
copyCodecs: transcodeType !== 'quick-transcode'
|
||||||
})
|
})
|
||||||
createHlsJobIfEnabled(originalFileHLSPayload)
|
await createHlsJobIfEnabled(user, originalFileHLSPayload)
|
||||||
|
|
||||||
const hasNewResolutions = createLowerResolutionsJobs(videoDatabase, videoFileResolution, isPortraitMode)
|
const hasNewResolutions = createLowerResolutionsJobs(videoDatabase, user, videoFileResolution, isPortraitMode)
|
||||||
|
|
||||||
if (!hasNewResolutions) {
|
if (!hasNewResolutions) {
|
||||||
// No transcoding to do, it's now published
|
// No transcoding to do, it's now published
|
||||||
|
@ -178,11 +183,12 @@ async function onVideoFileOptimizer (
|
||||||
|
|
||||||
async function onNewWebTorrentFileResolution (
|
async function onNewWebTorrentFileResolution (
|
||||||
video: MVideoUUID,
|
video: MVideoUUID,
|
||||||
|
user: MUserId,
|
||||||
payload?: NewResolutionTranscodingPayload | MergeAudioTranscodingPayload
|
payload?: NewResolutionTranscodingPayload | MergeAudioTranscodingPayload
|
||||||
) {
|
) {
|
||||||
await publishAndFederateIfNeeded(video)
|
await publishAndFederateIfNeeded(video)
|
||||||
|
|
||||||
createHlsJobIfEnabled(Object.assign({}, payload, { copyCodecs: true }))
|
await createHlsJobIfEnabled(user, Object.assign({}, payload, { copyCodecs: true }))
|
||||||
}
|
}
|
||||||
|
|
||||||
// ---------------------------------------------------------------------------
|
// ---------------------------------------------------------------------------
|
||||||
|
@ -194,9 +200,18 @@ export {
|
||||||
|
|
||||||
// ---------------------------------------------------------------------------
|
// ---------------------------------------------------------------------------
|
||||||
|
|
||||||
function createHlsJobIfEnabled (payload: { videoUUID: string, resolution: number, isPortraitMode?: boolean, copyCodecs: boolean }) {
|
async function createHlsJobIfEnabled (user: MUserId, payload: {
|
||||||
// Generate HLS playlist?
|
videoUUID: string
|
||||||
if (payload && CONFIG.TRANSCODING.HLS.ENABLED) {
|
resolution: number
|
||||||
|
isPortraitMode?: boolean
|
||||||
|
copyCodecs: boolean
|
||||||
|
}) {
|
||||||
|
if (!payload || CONFIG.TRANSCODING.HLS.ENABLED !== true) return
|
||||||
|
|
||||||
|
const jobOptions = {
|
||||||
|
priority: JOB_PRIORITY.TRANSCODING.NEW_RESOLUTION + await getJobTranscodingPriorityMalus(user)
|
||||||
|
}
|
||||||
|
|
||||||
const hlsTranscodingPayload: HLSTranscodingPayload = {
|
const hlsTranscodingPayload: HLSTranscodingPayload = {
|
||||||
type: 'new-resolution-to-hls',
|
type: 'new-resolution-to-hls',
|
||||||
videoUUID: payload.videoUUID,
|
videoUUID: payload.videoUUID,
|
||||||
|
@ -205,11 +220,15 @@ function createHlsJobIfEnabled (payload: { videoUUID: string, resolution: number
|
||||||
copyCodecs: payload.copyCodecs
|
copyCodecs: payload.copyCodecs
|
||||||
}
|
}
|
||||||
|
|
||||||
return JobQueue.Instance.createJob({ type: 'video-transcoding', payload: hlsTranscodingPayload })
|
return JobQueue.Instance.createJobWithPromise({ type: 'video-transcoding', payload: hlsTranscodingPayload }, jobOptions)
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
|
||||||
function createLowerResolutionsJobs (video: MVideoFullLight, videoFileResolution: number, isPortraitMode: boolean) {
|
async function createLowerResolutionsJobs (
|
||||||
|
video: MVideoFullLight,
|
||||||
|
user: MUserId,
|
||||||
|
videoFileResolution: number,
|
||||||
|
isPortraitMode: boolean
|
||||||
|
) {
|
||||||
// Create transcoding jobs if there are enabled resolutions
|
// Create transcoding jobs if there are enabled resolutions
|
||||||
const resolutionsEnabled = computeResolutionsToTranscode(videoFileResolution, 'vod')
|
const resolutionsEnabled = computeResolutionsToTranscode(videoFileResolution, 'vod')
|
||||||
logger.info(
|
logger.info(
|
||||||
|
@ -244,7 +263,11 @@ function createLowerResolutionsJobs (video: MVideoFullLight, videoFileResolution
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
JobQueue.Instance.createJob({ type: 'video-transcoding', payload: dataInput })
|
const jobOptions = {
|
||||||
|
priority: JOB_PRIORITY.TRANSCODING.NEW_RESOLUTION + await getJobTranscodingPriorityMalus(user)
|
||||||
|
}
|
||||||
|
|
||||||
|
JobQueue.Instance.createJob({ type: 'video-transcoding', payload: dataInput }, jobOptions)
|
||||||
}
|
}
|
||||||
|
|
||||||
logger.info('Transcoding jobs created for uuid %s.', video.uuid, { resolutionsEnabled })
|
logger.info('Transcoding jobs created for uuid %s.', video.uuid, { resolutionsEnabled })
|
||||||
|
|
|
@ -47,6 +47,7 @@ type CreateJobArgument =
|
||||||
|
|
||||||
type CreateJobOptions = {
|
type CreateJobOptions = {
|
||||||
delay?: number
|
delay?: number
|
||||||
|
priority?: number
|
||||||
}
|
}
|
||||||
|
|
||||||
const handlers: { [id in JobType]: (job: Bull.Job) => Promise<any> } = {
|
const handlers: { [id in JobType]: (job: Bull.Job) => Promise<any> } = {
|
||||||
|
@ -148,6 +149,7 @@ class JobQueue {
|
||||||
backoff: { delay: 60 * 1000, type: 'exponential' },
|
backoff: { delay: 60 * 1000, type: 'exponential' },
|
||||||
attempts: JOB_ATTEMPTS[obj.type],
|
attempts: JOB_ATTEMPTS[obj.type],
|
||||||
timeout: JOB_TTL[obj.type],
|
timeout: JOB_TTL[obj.type],
|
||||||
|
priority: options.priority,
|
||||||
delay: options.delay
|
delay: options.delay
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
|
@ -1,11 +1,13 @@
|
||||||
import { Transaction } from 'sequelize/types'
|
import { Transaction } from 'sequelize/types'
|
||||||
|
import { DEFAULT_AUDIO_RESOLUTION, JOB_PRIORITY } from '@server/initializers/constants'
|
||||||
import { sequelizeTypescript } from '@server/initializers/database'
|
import { sequelizeTypescript } from '@server/initializers/database'
|
||||||
import { TagModel } from '@server/models/video/tag'
|
import { TagModel } from '@server/models/video/tag'
|
||||||
import { VideoModel } from '@server/models/video/video'
|
import { VideoModel } from '@server/models/video/video'
|
||||||
import { FilteredModelAttributes } from '@server/types'
|
import { FilteredModelAttributes } from '@server/types'
|
||||||
import { MTag, MThumbnail, MVideoTag, MVideoThumbnail, MVideoUUID } from '@server/types/models'
|
import { MTag, MThumbnail, MUserId, MVideo, MVideoFile, MVideoTag, MVideoThumbnail, MVideoUUID } from '@server/types/models'
|
||||||
import { ThumbnailType, VideoCreate, VideoPrivacy } from '@shared/models'
|
import { ThumbnailType, VideoCreate, VideoPrivacy, VideoTranscodingPayload } from '@shared/models'
|
||||||
import { federateVideoIfNeeded } from './activitypub/videos'
|
import { federateVideoIfNeeded } from './activitypub/videos'
|
||||||
|
import { JobQueue } from './job-queue/job-queue'
|
||||||
import { Notifier } from './notifier'
|
import { Notifier } from './notifier'
|
||||||
import { createVideoMiniatureFromExisting } from './thumbnail'
|
import { createVideoMiniatureFromExisting } from './thumbnail'
|
||||||
|
|
||||||
|
@ -104,11 +106,47 @@ async function publishAndFederateIfNeeded (video: MVideoUUID, wasLive = false) {
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
async function addOptimizeOrMergeAudioJob (video: MVideo, videoFile: MVideoFile, user: MUserId) {
|
||||||
|
let dataInput: VideoTranscodingPayload
|
||||||
|
|
||||||
|
if (videoFile.isAudio()) {
|
||||||
|
dataInput = {
|
||||||
|
type: 'merge-audio-to-webtorrent',
|
||||||
|
resolution: DEFAULT_AUDIO_RESOLUTION,
|
||||||
|
videoUUID: video.uuid,
|
||||||
|
isNewVideo: true
|
||||||
|
}
|
||||||
|
} else {
|
||||||
|
dataInput = {
|
||||||
|
type: 'optimize-to-webtorrent',
|
||||||
|
videoUUID: video.uuid,
|
||||||
|
isNewVideo: true
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
const jobOptions = {
|
||||||
|
priority: JOB_PRIORITY.TRANSCODING.OPTIMIZER + await getJobTranscodingPriorityMalus(user)
|
||||||
|
}
|
||||||
|
|
||||||
|
return JobQueue.Instance.createJobWithPromise({ type: 'video-transcoding', payload: dataInput }, jobOptions)
|
||||||
|
}
|
||||||
|
|
||||||
|
async function getJobTranscodingPriorityMalus (user: MUserId) {
|
||||||
|
const now = new Date()
|
||||||
|
const lastWeek = new Date(now.getFullYear(), now.getMonth(), now.getDate() - 7)
|
||||||
|
|
||||||
|
const videoUploadedByUser = await VideoModel.countVideosUploadedByUserSince(user.id, lastWeek)
|
||||||
|
|
||||||
|
return videoUploadedByUser
|
||||||
|
}
|
||||||
|
|
||||||
// ---------------------------------------------------------------------------
|
// ---------------------------------------------------------------------------
|
||||||
|
|
||||||
export {
|
export {
|
||||||
buildLocalVideoFromReq,
|
buildLocalVideoFromReq,
|
||||||
publishAndFederateIfNeeded,
|
publishAndFederateIfNeeded,
|
||||||
buildVideoThumbnailsFromReq,
|
buildVideoThumbnailsFromReq,
|
||||||
setVideoTags
|
setVideoTags,
|
||||||
|
addOptimizeOrMergeAudioJob,
|
||||||
|
getJobTranscodingPriorityMalus
|
||||||
}
|
}
|
||||||
|
|
|
@ -129,6 +129,7 @@ import { VideoShareModel } from './video-share'
|
||||||
import { VideoStreamingPlaylistModel } from './video-streaming-playlist'
|
import { VideoStreamingPlaylistModel } from './video-streaming-playlist'
|
||||||
import { VideoTagModel } from './video-tag'
|
import { VideoTagModel } from './video-tag'
|
||||||
import { VideoViewModel } from './video-view'
|
import { VideoViewModel } from './video-view'
|
||||||
|
import { UserModel } from '../account/user'
|
||||||
|
|
||||||
export enum ScopeNames {
|
export enum ScopeNames {
|
||||||
AVAILABLE_FOR_LIST_IDS = 'AVAILABLE_FOR_LIST_IDS',
|
AVAILABLE_FOR_LIST_IDS = 'AVAILABLE_FOR_LIST_IDS',
|
||||||
|
@ -1198,6 +1199,39 @@ export class VideoModel extends Model {
|
||||||
return VideoModel.count(options)
|
return VideoModel.count(options)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
static countVideosUploadedByUserSince (userId: number, since: Date) {
|
||||||
|
const options = {
|
||||||
|
include: [
|
||||||
|
{
|
||||||
|
model: VideoChannelModel.unscoped(),
|
||||||
|
required: true,
|
||||||
|
include: [
|
||||||
|
{
|
||||||
|
model: AccountModel.unscoped(),
|
||||||
|
required: true,
|
||||||
|
include: [
|
||||||
|
{
|
||||||
|
model: UserModel.unscoped(),
|
||||||
|
required: true,
|
||||||
|
where: {
|
||||||
|
id: userId
|
||||||
|
}
|
||||||
|
}
|
||||||
|
]
|
||||||
|
}
|
||||||
|
]
|
||||||
|
}
|
||||||
|
],
|
||||||
|
where: {
|
||||||
|
createdAt: {
|
||||||
|
[Op.gte]: since
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
return VideoModel.unscoped().count(options)
|
||||||
|
}
|
||||||
|
|
||||||
static countLivesOfAccount (accountId: number) {
|
static countLivesOfAccount (accountId: number) {
|
||||||
const options = {
|
const options = {
|
||||||
where: {
|
where: {
|
||||||
|
|
|
@ -23,6 +23,7 @@ export interface Job {
|
||||||
state: JobState
|
state: JobState
|
||||||
type: JobType
|
type: JobType
|
||||||
data: any
|
data: any
|
||||||
|
priority: number
|
||||||
progress: number
|
progress: number
|
||||||
error: any
|
error: any
|
||||||
createdAt: Date | string
|
createdAt: Date | string
|
||||||
|
|
Loading…
Reference in a new issue