[Feature] Global Job Concurrency

[auanasgheps]

The feature

On poor systems like mine (i3-8300, 4 cores) Immich overwhelms the system when I ingest a lot of media.

I have observed that if I pause ALL services and then have only one running at the time, it is much better: sure it may take more time, but at least won't hog down the CPU to 99.99%, and the system remains usable to a certain degree.

So please, introduce a "Global Job Concurrency", and not only for a specific job, so I can set it to 1 and not have my server die under Immich processing!

Platform

• Server
• Web
• Mobile

[AngelaDMerkel]

I agree with this:

Assuming I have 8 cores, I might limit the microservice to four so transcoding doesn't take forever but then when I add images the CLIP or facial recognition might bring the system to full utilisation. It would be helpful to assign a total system utilisation for the entire stack and then allow concurrency to provide full resources to each process sequentially, rather than hope a particular upload doesn't allow various containers in the stack to stress the system

[jamess60]

+1

Running Immich with docker in a 12 core VM (2x Xeon E5 2697 v2 system). I have 20 thumbnail jobs running at once with no other tasks (manually paused them). CPU spikes 50-100. System becomes unusable if I have any more running.

[jamess60]

My only dev skillset is some limited python, but surely it cant be much harder than adding the total of the pre-existing concurrences together, if that value is greater than global limit, throttle some back. You could even enter a while true loop and just pause one entire task if thats easier. Theres already a pause button for tasks so there must be a function for it.

In python, I doubt that would be more than maybe 5-10 lines of code.

[jrasm91]

There is no easy way to set a global limit unfortunately

[hirenshah]

Rather than setting an overall limit, would it be possible to run each job sequentially rather than in parallel?

[mertalev]

The main issue with setting a global limit (or making jobs run sequentially) is that the job scheduler we use, BullMQ, handles concurrency at the queue level. Since the scheduler is in charge of handling concurrency, we don't have a non-hacky way of adding a global limit. This is a feature in BullMQ Pro, so it's also unlikely it will be added to BullMQ in the future.

We intend to switch to a different job scheduler at some point (not for this, but to drop Redis and handle jobs in Postgres). It may be more feasible at that point, but it again depends on what the scheduler supports.

[yuhuan417]

bullmq global concurrency: taskforcesh/bullmq#2496

[Marco-Z]

Assuming you are running with docker compose, you can limit the resources on a container level. See https://docs.docker.com/compose/compose-file/deploy/#resources
For example on my Raspberry Pi 4 (4GB), I run the immich-microservices with:

deploy:
  resources:
    limits:
      cpus: 2
      memory: 2500mb

You can do the same on the immich-machine-learning (or even offload that to a remote machine when ingesting many files).

[auanasgheps]

Thank you! This is the closest we can get to a solution! I will limit the microservices container to 2 cores as well, since I have 4.

[jamess60]

This is a big help in the sense that it stops the other containers suffering / docker host becoming unstable, but then it does mean immich cant really stretch its legs and utilize the full hardware potential either.

Going to implement this in my setup for now as a fail safe

[yydcool]

+1
I'm using a low-memory machine and I already set concurrency to 1 for every job. But I still see logs showing "Worker (pid:104) was sent SIGKILL! Perhaps out of memory?" from the immich_machine_learning container.

It turns out 3 machine learning jobs [smart search] [face detect] [face recognition] will be running at the same time.

I have to pause these jobs, and manually un-pause them one at a time:

It seems BullMQ made this complicated. Would it make sense to add a CLI feature to get the each job queue status and pause/un-pause each job queue?

With that, I would be able to write our own script to control the global job concurrency.

[yydcool]

wow that's great, thanks for the information!

[yydcool]

I've created a script to set only one job queue to be active, now effectively global concurrency = 1
https://gist.github.com/yydcool/5df3ce4a86e7b1931e794f4e6095d322

[ppskj178]

하나의 작업 대기열만 활성화하도록 설정하는 스크립트를 만들었고, 이제 효과적으로 전역 동시성 = 1이 되었습니다. https://gist.github.com/yydcool/5df3ce4a86e7b1931e794f4e6095d322

How do I apply it?

Apart from this, this feature should be supported by immich itself. For most people, creating a preview is important and AI learning can be done slowly.

[WINVIST]

I've created a script to set only one job queue to be active, now effectively global concurrency = 1 https://gist.github.com/yydcool/5df3ce4a86e7b1931e794f4e6095d322

Hi, saw your script. I want to know if you have made any changes to it? I've had it for more than 24 hours without completing a single task. I can't figure out what's wrong

[DidouJazz]

Hey, here is the one I am using (with good success so far) !

# /usr/local/bin/python3.9 /volume2/docker/Immich/API/main.py

import requests
import json
import time

# Configuration for the API endpoint and headers
url = "http://127.0.0.1:2283/api/jobs"
headers = {
    'Content-Type': 'application/json',
    'Accept': 'application/json',
    'x-api-key': 'XXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXX'
}

def fetch_jobs():
    """
    Fetch the current job statuses from the API.

    Returns:
        dict: A dictionary of job information, or an empty dict on error.
    """
    try:
        response = requests.get(url, headers=headers)
        response.raise_for_status()
        return response.json()
    except (requests.RequestException, json.JSONDecodeError) as e:
        print(f"Error fetching jobs: {e}")
        return {}

def set_job(job, command):
    """
    Send a command to pause or resume a specific job via the API.

    Args:
        job (str): The job name.
        command (str): The command to send ("pause" or "resume").
    """
    try:
        payload = json.dumps({"command": command, "force": False})
        response = requests.put(f"{url}/{job}", headers=headers, data=payload)
        response.raise_for_status()
        # print(f"Set job {job} to {command}")
    except requests.RequestException as e:
        print(f"Error setting job {job} to {command}: {e}")

def manage_jobs():
    """
    Main logic to manage jobs:
    - Fetches all jobs from the API.
    - Finds the first job in the managed list that has any activity.
    - Unpauses that job and pauses all others in the managed list.
    """
    jobs = fetch_jobs()
    if not jobs:
        return

    # List of jobs to manage in priority order
    managed_job_list = [
        "metadataExtraction",
        "storageTemplateMigration",
        "thumbnailGeneration",
        "smartSearch",
        "duplicateDetection",
        "faceDetection",
        "facialRecognition",
        "videoConversion"
    ]

    # Find the first job with any jobCounts > 0 (active, waiting, paused, or delayed)
    job_to_unpause = None
    for job in managed_job_list:
        job_info = jobs.get(job)
        if job_info and (
            job_info['jobCounts']['active'] > 0 or
            job_info['jobCounts']['waiting'] > 0 or
            job_info['jobCounts']['paused'] > 0 or
            job_info['jobCounts']['delayed'] > 0
        ):
            job_to_unpause = job
            break

    # Unpause the first found job, pause all others in managed_job_list
    for job in managed_job_list:
        if job == job_to_unpause:
            # print(f"Unpausing job: {job}")
            set_job(job, "resume")
        else:
            # print(f"Pausing job: {job}")
            set_job(job, "pause")

if __name__ == "__main__":
    # Run the job manager loop every 10 seconds
    while True:
        manage_jobs()
        time.sleep(10)