aboutsummaryrefslogtreecommitdiffhomepage
path: root/server/lib/jobs
diff options
context:
space:
mode:
Diffstat (limited to 'server/lib/jobs')
-rw-r--r--server/lib/jobs/handlers/video-file-optimizer.ts88
-rw-r--r--server/lib/jobs/handlers/video-file-transcoder.ts21
-rw-r--r--server/lib/jobs/job-scheduler.ts118
3 files changed, 115 insertions, 112 deletions
diff --git a/server/lib/jobs/handlers/video-file-optimizer.ts b/server/lib/jobs/handlers/video-file-optimizer.ts
index 63a51064c..799ba8b01 100644
--- a/server/lib/jobs/handlers/video-file-optimizer.ts
+++ b/server/lib/jobs/handlers/video-file-optimizer.ts
@@ -1,4 +1,4 @@
1import * as Promise from 'bluebird' 1import * as Bluebird from 'bluebird'
2 2
3import { database as db } from '../../../initializers/database' 3import { database as db } from '../../../initializers/database'
4import { logger, computeResolutionsToTranscode } from '../../../helpers' 4import { logger, computeResolutionsToTranscode } from '../../../helpers'
@@ -6,16 +6,17 @@ import { VideoInstance } from '../../../models'
6import { addVideoToFriends } from '../../friends' 6import { addVideoToFriends } from '../../friends'
7import { JobScheduler } from '../job-scheduler' 7import { JobScheduler } from '../job-scheduler'
8 8
9function process (data: { videoUUID: string }, jobId: number) { 9async function process (data: { videoUUID: string }, jobId: number) {
10 return db.Video.loadByUUIDAndPopulateAuthorAndPodAndTags(data.videoUUID).then(video => { 10 const video = await db.Video.loadByUUIDAndPopulateAuthorAndPodAndTags(data.videoUUID)
11 // No video, maybe deleted? 11 // No video, maybe deleted?
12 if (!video) { 12 if (!video) {
13 logger.info('Do not process job %d, video does not exist.', jobId, { videoUUID: video.uuid }) 13 logger.info('Do not process job %d, video does not exist.', jobId, { videoUUID: video.uuid })
14 return undefined 14 return undefined
15 } 15 }
16
17 await video.optimizeOriginalVideofile()
16 18
17 return video.optimizeOriginalVideofile().then(() => video) 19 return video
18 })
19} 20}
20 21
21function onError (err: Error, jobId: number) { 22function onError (err: Error, jobId: number) {
@@ -23,33 +24,31 @@ function onError (err: Error, jobId: number) {
23 return Promise.resolve() 24 return Promise.resolve()
24} 25}
25 26
26function onSuccess (jobId: number, video: VideoInstance) { 27async function onSuccess (jobId: number, video: VideoInstance) {
27 if (video === undefined) return undefined 28 if (video === undefined) return undefined
28 29
29 logger.info('Job %d is a success.', jobId) 30 logger.info('Job %d is a success.', jobId)
30 31
31 video.toAddRemoteJSON() 32 const remoteVideo = await video.toAddRemoteJSON()
32 .then(remoteVideo => { 33
33 // Now we'll add the video's meta data to our friends 34 // Now we'll add the video's meta data to our friends
34 return addVideoToFriends(remoteVideo, null) 35 await addVideoToFriends(remoteVideo, null)
35 }) 36
36 .then(() => { 37 const originalFileHeight = await video.getOriginalFileHeight()
37 return video.getOriginalFileHeight() 38 // Create transcoding jobs if there are enabled resolutions
38 }) 39
39 .then(originalFileHeight => { 40 const resolutionsEnabled = computeResolutionsToTranscode(originalFileHeight)
40 // Create transcoding jobs if there are enabled resolutions 41 logger.info(
41 const resolutionsEnabled = computeResolutionsToTranscode(originalFileHeight) 42 'Resolutions computed for video %s and origin file height of %d.', video.uuid, originalFileHeight,
42 logger.info( 43 { resolutions: resolutionsEnabled }
43 'Resolutions computed for video %s and origin file height of %d.', video.uuid, originalFileHeight, 44 )
44 { resolutions: resolutionsEnabled } 45
45 ) 46 if (resolutionsEnabled.length !== 0) {
46 47 try {
47 if (resolutionsEnabled.length === 0) return undefined 48 await db.sequelize.transaction(async t => {
48 49 const tasks: Bluebird<any>[] = []
49 return db.sequelize.transaction(t => { 50
50 const tasks: Promise<any>[] = [] 51 for (const resolution of resolutionsEnabled) {
51
52 resolutionsEnabled.forEach(resolution => {
53 const dataInput = { 52 const dataInput = {
54 videoUUID: video.uuid, 53 videoUUID: video.uuid,
55 resolution 54 resolution
@@ -57,24 +56,19 @@ function onSuccess (jobId: number, video: VideoInstance) {
57 56
58 const p = JobScheduler.Instance.createJob(t, 'videoFileTranscoder', dataInput) 57 const p = JobScheduler.Instance.createJob(t, 'videoFileTranscoder', dataInput)
59 tasks.push(p) 58 tasks.push(p)
60 }) 59 }
61 60
62 return Promise.all(tasks).then(() => resolutionsEnabled) 61 await Promise.all(tasks)
63 }) 62 })
64 })
65 .then(resolutionsEnabled => {
66 if (resolutionsEnabled === undefined) {
67 logger.info('No transcoding jobs created for video %s (no resolutions enabled).')
68 return undefined
69 }
70 63
71 logger.info('Transcoding jobs created for uuid %s.', video.uuid, { resolutionsEnabled }) 64 logger.info('Transcoding jobs created for uuid %s.', video.uuid, { resolutionsEnabled })
72 }) 65 } catch (err) {
73 .catch((err: Error) => { 66 logger.warn('Cannot transcode the video.', err)
74 logger.debug('Cannot transcode the video.', err) 67 }
75 throw err 68 } else {
76 }) 69 logger.info('No transcoding jobs created for video %s (no resolutions enabled).')
77 70 return undefined
71 }
78} 72}
79 73
80// --------------------------------------------------------------------------- 74// ---------------------------------------------------------------------------
diff --git a/server/lib/jobs/handlers/video-file-transcoder.ts b/server/lib/jobs/handlers/video-file-transcoder.ts
index 0dafee566..b240ff58a 100644
--- a/server/lib/jobs/handlers/video-file-transcoder.ts
+++ b/server/lib/jobs/handlers/video-file-transcoder.ts
@@ -4,16 +4,17 @@ import { logger } from '../../../helpers'
4import { VideoInstance } from '../../../models' 4import { VideoInstance } from '../../../models'
5import { VideoResolution } from '../../../../shared' 5import { VideoResolution } from '../../../../shared'
6 6
7function process (data: { videoUUID: string, resolution: VideoResolution }, jobId: number) { 7async function process (data: { videoUUID: string, resolution: VideoResolution }, jobId: number) {
8 return db.Video.loadByUUIDAndPopulateAuthorAndPodAndTags(data.videoUUID).then(video => { 8 const video = await db.Video.loadByUUIDAndPopulateAuthorAndPodAndTags(data.videoUUID)
9 // No video, maybe deleted? 9 // No video, maybe deleted?
10 if (!video) { 10 if (!video) {
11 logger.info('Do not process job %d, video does not exist.', jobId, { videoUUID: video.uuid }) 11 logger.info('Do not process job %d, video does not exist.', jobId, { videoUUID: video.uuid })
12 return undefined 12 return undefined
13 } 13 }
14 14
15 return video.transcodeOriginalVideofile(data.resolution).then(() => video) 15 await video.transcodeOriginalVideofile(data.resolution)
16 }) 16
17 return video
17} 18}
18 19
19function onError (err: Error, jobId: number) { 20function onError (err: Error, jobId: number) {
diff --git a/server/lib/jobs/job-scheduler.ts b/server/lib/jobs/job-scheduler.ts
index c2409d20c..61d483268 100644
--- a/server/lib/jobs/job-scheduler.ts
+++ b/server/lib/jobs/job-scheduler.ts
@@ -23,7 +23,7 @@ class JobScheduler {
23 return this.instance || (this.instance = new this()) 23 return this.instance || (this.instance = new this())
24 } 24 }
25 25
26 activate () { 26 async activate () {
27 const limit = JOBS_FETCH_LIMIT_PER_CYCLE 27 const limit = JOBS_FETCH_LIMIT_PER_CYCLE
28 28
29 logger.info('Jobs scheduler activated.') 29 logger.info('Jobs scheduler activated.')
@@ -32,32 +32,36 @@ class JobScheduler {
32 32
33 // Finish processing jobs from a previous start 33 // Finish processing jobs from a previous start
34 const state = JOB_STATES.PROCESSING 34 const state = JOB_STATES.PROCESSING
35 db.Job.listWithLimit(limit, state) 35 try {
36 .then(jobs => { 36 const jobs = await db.Job.listWithLimit(limit, state)
37 this.enqueueJobs(jobsQueue, jobs) 37
38 38 this.enqueueJobs(jobsQueue, jobs)
39 forever( 39 } catch (err) {
40 next => { 40 logger.error('Cannot list pending jobs.', err)
41 if (jobsQueue.length() !== 0) { 41 }
42 // Finish processing the queue first 42
43 return setTimeout(next, JOBS_FETCHING_INTERVAL) 43 forever(
44 } 44 async next => {
45 45 if (jobsQueue.length() !== 0) {
46 const state = JOB_STATES.PENDING 46 // Finish processing the queue first
47 db.Job.listWithLimit(limit, state) 47 return setTimeout(next, JOBS_FETCHING_INTERVAL)
48 .then(jobs => { 48 }
49 this.enqueueJobs(jobsQueue, jobs) 49
50 50 const state = JOB_STATES.PENDING
51 // Optimization: we could use "drain" from queue object 51 try {
52 return setTimeout(next, JOBS_FETCHING_INTERVAL) 52 const jobs = await db.Job.listWithLimit(limit, state)
53 }) 53
54 .catch(err => logger.error('Cannot list pending jobs.', err)) 54 this.enqueueJobs(jobsQueue, jobs)
55 }, 55 } catch (err) {
56 56 logger.error('Cannot list pending jobs.', err)
57 err => logger.error('Error in job scheduler queue.', err) 57 }
58 ) 58
59 }) 59 // Optimization: we could use "drain" from queue object
60 .catch(err => logger.error('Cannot list pending jobs.', err)) 60 return setTimeout(next, JOBS_FETCHING_INTERVAL)
61 },
62
63 err => logger.error('Error in job scheduler queue.', err)
64 )
61 } 65 }
62 66
63 createJob (transaction: Sequelize.Transaction, handlerName: string, handlerInputData: object) { 67 createJob (transaction: Sequelize.Transaction, handlerName: string, handlerInputData: object) {
@@ -75,7 +79,7 @@ class JobScheduler {
75 jobs.forEach(job => jobsQueue.push(job)) 79 jobs.forEach(job => jobsQueue.push(job))
76 } 80 }
77 81
78 private processJob (job: JobInstance, callback: (err: Error) => void) { 82 private async processJob (job: JobInstance, callback: (err: Error) => void) {
79 const jobHandler = jobHandlers[job.handlerName] 83 const jobHandler = jobHandlers[job.handlerName]
80 if (jobHandler === undefined) { 84 if (jobHandler === undefined) {
81 logger.error('Unknown job handler for job %s.', job.handlerName) 85 logger.error('Unknown job handler for job %s.', job.handlerName)
@@ -85,41 +89,45 @@ class JobScheduler {
85 logger.info('Processing job %d with handler %s.', job.id, job.handlerName) 89 logger.info('Processing job %d with handler %s.', job.id, job.handlerName)
86 90
87 job.state = JOB_STATES.PROCESSING 91 job.state = JOB_STATES.PROCESSING
88 return job.save() 92 await job.save()
89 .then(() => { 93
90 return jobHandler.process(job.handlerInputData, job.id) 94 try {
91 }) 95 const result = await jobHandler.process(job.handlerInputData, job.id)
92 .then( 96 await this.onJobSuccess(jobHandler, job, result)
93 result => { 97 } catch (err) {
94 return this.onJobSuccess(jobHandler, job, result) 98 logger.error('Error in job handler %s.', job.handlerName, err)
95 }, 99
96 100 try {
97 err => { 101 await this.onJobError(jobHandler, job, err)
98 logger.error('Error in job handler %s.', job.handlerName, err) 102 } catch (innerErr) {
99 return this.onJobError(jobHandler, job, err) 103 this.cannotSaveJobError(innerErr)
100 } 104 return callback(innerErr)
101 ) 105 }
102 .then(() => callback(null)) 106 }
103 .catch(err => { 107
104 this.cannotSaveJobError(err) 108 callback(null)
105 return callback(err)
106 })
107 } 109 }
108 110
109 private onJobError (jobHandler: JobHandler<any>, job: JobInstance, err: Error) { 111 private async onJobError (jobHandler: JobHandler<any>, job: JobInstance, err: Error) {
110 job.state = JOB_STATES.ERROR 112 job.state = JOB_STATES.ERROR
111 113
112 return job.save() 114 try {
113 .then(() => jobHandler.onError(err, job.id)) 115 await job.save()
114 .catch(err => this.cannotSaveJobError(err)) 116 await jobHandler.onError(err, job.id)
117 } catch (err) {
118 this.cannotSaveJobError(err)
119 }
115 } 120 }
116 121
117 private onJobSuccess (jobHandler: JobHandler<any>, job: JobInstance, jobResult: any) { 122 private async onJobSuccess (jobHandler: JobHandler<any>, job: JobInstance, jobResult: any) {
118 job.state = JOB_STATES.SUCCESS 123 job.state = JOB_STATES.SUCCESS
119 124
120 return job.save() 125 try {
121 .then(() => jobHandler.onSuccess(job.id, jobResult)) 126 await job.save()
122 .catch(err => this.cannotSaveJobError(err)) 127 jobHandler.onSuccess(job.id, jobResult)
128 } catch (err) {
129 this.cannotSaveJobError(err)
130 }
123 } 131 }
124 132
125 private cannotSaveJobError (err: Error) { 133 private cannotSaveJobError (err: Error) {