@@ -72,6 +72,7 @@ export class C2DEngineDocker extends C2DEngine {
7272 private cpuAllocations : Map < string , number [ ] > = new Map ( )
7373 private envCpuCores : number [ ] = [ ]
7474 private cpuOffset : number
75+
7576 public constructor (
7677 clusterConfig : C2DClusterInfo ,
7778 db : C2DDatabase ,
@@ -414,11 +415,11 @@ export class C2DEngineDocker extends C2DEngine {
414415 }
415416
416417 // Process each job to determine what operation is needed
418+ let duration
417419 for ( const job of jobs ) {
418420 // Calculate algo duration
419- const algoDuration =
420- parseFloat ( job . algoStopTimestamp ) - parseFloat ( job . algoStartTimestamp )
421- job . algoDuration = algoDuration
421+ duration = parseFloat ( job . algoStopTimestamp ) - parseFloat ( job . algoStartTimestamp )
422+ duration += this . getValidBuildDurationSeconds ( job )
422423
423424 // Free jobs or jobs without payment info - mark as finished
424425 if ( job . isFree || ! job . payment ) {
@@ -455,7 +456,7 @@ export class C2DEngineDocker extends C2DEngine {
455456 continue
456457 }
457458
458- let minDuration = Math . abs ( algoDuration )
459+ let minDuration = Math . abs ( duration )
459460 if ( minDuration > job . maxJobDuration ) {
460461 minDuration = job . maxJobDuration
461462 }
@@ -1107,6 +1108,9 @@ export class C2DEngineDocker extends C2DEngine {
11071108 throw new Error ( `additionalDockerFiles cannot be used with queued jobs` )
11081109 }
11091110 }
1111+ // if (algorithm.meta.container && algorithm.meta.container.dockerfile && isFree) {
1112+ // throw new Error(`Building image is not allowed for free jobs`)
1113+ // }
11101114
11111115 const job : DBComputeJob = {
11121116 clusterHash : this . getC2DConfig ( ) . hash ,
@@ -1147,7 +1151,9 @@ export class C2DEngineDocker extends C2DEngine {
11471151 algoDuration : 0 ,
11481152 queueMaxWaitTime : queueMaxWaitTime || 0 ,
11491153 encryptedDockerRegistryAuth, // we store the encrypted docker registry auth in the job
1150- output
1154+ output,
1155+ buildStartTimestamp : '0' ,
1156+ buildStopTimestamp : '0'
11511157 }
11521158
11531159 if ( algorithm . meta . container && algorithm . meta . container . dockerfile ) {
@@ -1606,6 +1612,19 @@ export class C2DEngineDocker extends C2DEngine {
16061612 }
16071613
16081614 if ( job . status === C2DStatusNumber . ConfiguringVolumes ) {
1615+ // we have the image (etiher pulled or built)
1616+ // if built, check if build process took all allocated time
1617+ // if yes, stop the job
1618+ const buildDuration = this . getValidBuildDurationSeconds ( job )
1619+ if ( buildDuration > 0 && buildDuration >= job . maxJobDuration ) {
1620+ job . isStarted = false
1621+ job . status = C2DStatusNumber . PublishingResults
1622+ job . statusText = C2DStatusText . PublishingResults
1623+ job . algoStartTimestamp = '0'
1624+ job . algoStopTimestamp = '0'
1625+ job . isRunning = false
1626+ await this . db . updateJob ( job )
1627+ }
16091628 // create the volume & create container
16101629 // TO DO C2D: Choose driver & size
16111630 // get env info
@@ -1814,7 +1833,13 @@ export class C2DEngineDocker extends C2DEngine {
18141833 }
18151834
18161835 const timeNow = Date . now ( ) / 1000
1817- const expiry = parseFloat ( job . algoStartTimestamp ) + job . maxJobDuration
1836+ let expiry
1837+
1838+ const buildDuration = this . getValidBuildDurationSeconds ( job )
1839+ if ( buildDuration > 0 ) {
1840+ // if job has build time, reduce the remaining algorithm runtime budget
1841+ expiry = parseFloat ( job . algoStartTimestamp ) + job . maxJobDuration - buildDuration
1842+ } else expiry = parseFloat ( job . algoStartTimestamp ) + job . maxJobDuration
18181843 CORE_LOGGER . debug (
18191844 'container running since timeNow: ' + timeNow + ' , Expiry: ' + expiry
18201845 )
@@ -1964,6 +1989,14 @@ export class C2DEngineDocker extends C2DEngine {
19641989
19651990 private allocateCpus ( jobId : string , count : number ) : string | null {
19661991 if ( this . envCpuCores . length === 0 || count <= 0 ) return null
1992+ const existing = this . cpuAllocations . get ( jobId )
1993+ if ( existing && existing . length > 0 ) {
1994+ const cpusetStr = existing . join ( ',' )
1995+ CORE_LOGGER . info (
1996+ `CPU affinity: reusing existing cores [${ cpusetStr } ] for job ${ jobId } `
1997+ )
1998+ return cpusetStr
1999+ }
19672000
19682001 const usedCores = new Set < number > ( )
19692002 for ( const cores of this . cpuAllocations . values ( ) ) {
@@ -2341,7 +2374,7 @@ export class C2DEngineDocker extends C2DEngine {
23412374 const imageLogFile =
23422375 this . getC2DConfig ( ) . tempFolder + '/' + job . jobId + '/data/logs/image.log'
23432376 const controller = new AbortController ( )
2344- const timeoutMs = 5 * 60 * 1000
2377+ const timeoutMs = job . maxJobDuration * 1000
23452378 const timer = setTimeout ( ( ) => controller . abort ( ) , timeoutMs )
23462379 try {
23472380 const pack = tarStream . pack ( )
@@ -2355,18 +2388,29 @@ export class C2DEngineDocker extends C2DEngine {
23552388 }
23562389 }
23572390 pack . finalize ( )
2391+ job . buildStartTimestamp = String ( Date . now ( ) / 1000 )
2392+ await this . db . updateJob ( job )
23582393
2359- // Build the image using the tar stream as context (Node IncomingMessage extends stream.Readable)
2360- const buildStream = ( await this . docker . buildImage ( pack , {
2394+ const cpuperiod = 100000
2395+ const ramGb = this . getResourceRequest ( job . resources , 'ram' )
2396+ const ramBytes =
2397+ ramGb && ramGb > 0 ? ramGb * 1024 * 1024 * 1024 : 1024 * 1024 * 1024
2398+
2399+ const cpus = this . getResourceRequest ( job . resources , 'cpu' )
2400+ const cpuquota = cpus && cpus > 0 ? Math . floor ( cpus * cpuperiod ) : 50000
2401+
2402+ const buildOptions : Dockerode . ImageBuildOptions = {
23612403 t : job . containerImage ,
2362- memory : 1024 * 1024 * 1024 , // 1GB RAM in bytes
2363- memswap : - 1 , // Disable swap
2364- cpushares : 512 , // CPU Shares (default is 1024)
2365- cpuquota : 50000 , // 50% of one CPU ( 100000 = 1 CPU)
2366- cpuperiod : 100000 , // Default period
2404+ memory : ramBytes ,
2405+ memswap : ramBytes , // same as memory => no swap
2406+ cpushares : 1024 , // CPU Shares (default is 1024)
2407+ cpuquota, // 100000 = 1 CPU with cpuperiod=100000
2408+ cpuperiod,
23672409 nocache : true , // prevent cache poison
23682410 abortSignal : controller . signal
2369- } ) ) as Readable
2411+ }
2412+ // Build the image using the tar stream as context (Node IncomingMessage extends stream.Readable)
2413+ const buildStream = ( await this . docker . buildImage ( pack , buildOptions ) ) as Readable
23702414
23712415 const onBuildData = ( data : Buffer ) => {
23722416 try {
@@ -2405,9 +2449,23 @@ export class C2DEngineDocker extends C2DEngine {
24052449 }
24062450 controller . signal . addEventListener ( 'abort' , onAbort , { once : true } )
24072451 const onSuccess = ( ) => {
2408- finish ( ( ) => {
2452+ finish ( async ( ) => {
24092453 detachBuildLog ( )
24102454 controller . signal . removeEventListener ( 'abort' , onAbort )
2455+
2456+ // Build stream completed, but does the image actually exist?
2457+ try {
2458+ await this . docker . getImage ( job . containerImage ) . inspect ( )
2459+ } catch ( e ) {
2460+ return reject (
2461+ new Error (
2462+ `Cannot find image '${ job . containerImage } ' after building. Most likely it failed: ${
2463+ ( e as Error ) ?. message || String ( e )
2464+ } `
2465+ )
2466+ )
2467+ }
2468+
24112469 CORE_LOGGER . debug ( `Image '${ job . containerImage } ' built successfully.` )
24122470 this . updateImageUsage ( job . containerImage ) . catch ( ( e ) => {
24132471 CORE_LOGGER . debug ( `Failed to track image usage: ${ e . message } ` )
@@ -2430,6 +2488,7 @@ export class C2DEngineDocker extends C2DEngine {
24302488 } )
24312489 job . status = C2DStatusNumber . ConfiguringVolumes
24322490 job . statusText = C2DStatusText . ConfiguringVolumes
2491+ job . buildStopTimestamp = String ( Date . now ( ) / 1000 )
24332492 await this . db . updateJob ( job )
24342493 } catch ( err ) {
24352494 const aborted =
@@ -2448,6 +2507,7 @@ export class C2DEngineDocker extends C2DEngine {
24482507 }
24492508 job . status = C2DStatusNumber . BuildImageFailed
24502509 job . statusText = C2DStatusText . BuildImageFailed
2510+ job . buildStopTimestamp = String ( Date . now ( ) / 1000 )
24512511 job . isRunning = false
24522512 job . dateFinished = String ( Date . now ( ) / 1000 )
24532513 await this . db . updateJob ( job )
@@ -2843,6 +2903,18 @@ export class C2DEngineDocker extends C2DEngine {
28432903 }
28442904 return false
28452905 }
2906+
2907+ private getValidBuildDurationSeconds ( job : DBComputeJob ) : number {
2908+ const startRaw = job . buildStartTimestamp
2909+ const stopRaw = job . buildStopTimestamp
2910+ if ( ! startRaw || ! stopRaw ) return 0
2911+ const start = Number . parseFloat ( startRaw )
2912+ const stop = Number . parseFloat ( stopRaw )
2913+ if ( ! Number . isFinite ( start ) || ! Number . isFinite ( stop ) ) return 0
2914+ if ( start <= 0 ) return 0
2915+ if ( stop < start ) return 0
2916+ return stop - start
2917+ }
28462918}
28472919
28482920// this uses the docker engine, but exposes only one env, the free one
0 commit comments