Defining Cloud Resources

​

Prespecified Task Resource

The Latch SDK currently supports a set of prespecified task resource requests represented as decorators:

• small_task: 2 cpus, 4 gigs of memory, 0 gpus
• medium_task: 32 cpus, 128 gigs of memory, 0 gpus
• large_task: 96 cpus, 192 gig sof memory, 0 gpus
• small_gpu_task: 8 cpus, 32 gigs of memory, 1 gpu (24 gigs of VRAM, 9,216 cuda cores)
• large_gpu_task: 31 cpus, 120 gigs of memory, 1 gpu (24 gigs of VRAM, 9,216 cuda cores)
• v100_x1_task: 16 cpus, 64 gigs of memory, 1 V100 gpu (16 gigs of VRAM, 5,120 cuda cores)
• v100_x4_task: 64 cpus, 256 gigs of memory, 4 V100 gpus (64 gigs of VRAM, 20,480 cuda cores)
• v100_x8_task: 128 cpus, 512 gigs of memory, 8 V100 gpus (128 gigs of VRAM, 40,960 cuda cores)
• g6e_xlarge_task: 4 cpus, 32 gigs of memory, 1 L40s gpu
• g6e_2xlarge_task: 8 cpus, 64 gigs of memory, 1 L40s gpu
• g6e_4xlarge_task: 16 cpus, 128 gigs of memory, 1 L40s gpu
• g6e_8xlarge_task: 32 cpus, 256 gigs of memory, 1 L40s gpu
• g6e_12xlarge_task: 48 cpus, 384 gigs of memory, 1 L40s gpu
• g6e_16xlarge_task: 64 cpus, 512 gigs of memory, 1 L40s gpu
• g6e_24xlarge_task: 96 cpus, 768 gigs of memory, 4 L40s gpus

We use the tasks as follows:

from latch.resources.tasks import small_task, large_gpu_task, v100_x1_task, g6e_xlarge_task

@small_task
def my_task(
    ...
):
    ...

@large_gpu_task
def inference(
    ...
):
    ...

​

Custom Task Resource

You can also arbitrarily specify task resources using @custom_task:

from latch import custom_task

@custom_task(cpu, memory) # cpu: int, memory: int
def my_task(
    ...
):
    ...

​

Dynamic Task Resource

You can dynamically define task resources based on the tasks’ input parameters by passing functions as arguments for the custom_task decorator. The provided functions will execute at runtime, and the task will launch with the resulting resource values:

...

from latch import custom_task
from latch.types.file import LatchFile

def allocate_cpu(files: List[LatchFile], **kwargs) -> int: # number of cores to allocate
    return min(8, len(files))

def allocate_storage(files: List[LatchFile], **kwargs) -> int: # GiBs of storage to allocate
    return sum([1.5 * file.size() for file in files]) // 1024**3

@custom_task(cpu=allocate_cpu, memory=8, storage=allocate_storage)
def my_task(files: List[LatchFile], count: int):
    ...

In the provided example, the allocate_cpu function is designed to process the input parameters files. Upon execution, the function returns an integer representing the total number of CPU cores that should be allocated to the task based on the input file size.