xqtor

Xqtor is an innovative, flexible alternative to the standard concurrent.futures.ThreadPoolExecutor in Python. Unlike the original ThreadPoolExecutor, which strictly schedules tasks based on a fixed number of workers, Xqtor offers a more dynamic approach by allowing tasks to specify the resources they require. This provides greater flexibility for task scheduling based on available resources.

Key Features

• Resource-Based Scheduling:

  • Set a total quantity of resources when initializing Xqtor.
  • Each submitted task can specify how many resources it needs, whether that's fractions or specific units.

• Integration with Existing Code:

  • The constructor parameters work just like in ThreadPoolExecutor (e.g., set max_workers or initializer), with the addition of the available parameter.

Quick Start Guide

To get started with Xqtor, install it via pip:

pip install xqtor

Basic Usage Example

Here's a sample code snippet demonstrating how to use Xqtor with different resource requirements:

from xqtor import Xqtor
import time

def task_4_cpu():
    print("running task_4_cpu")
    time.sleep(1)

def task_2_cpu():
    print("running task_2_cpu")
    time.sleep(1)

def task_half_cpu():
    print("running task_half_cpu")
    time.sleep(1)

executor = Xqtor(available=5.5)

futures = []
futures.append(executor.submit((task_4_cpu, 4.0)))
futures.append(executor.submit((task_2_cpu, 2.0)))
futures.append(executor.submit((task_half_cpu, 0.5)))

for future in futures:
    future.result()

The output of this example shows how tasks are managed based on the specified resources:

# T=0
running task_4_cpu
running task_half_cpu

# T=1
running task_2_cpu

Advanced Resource Management

For more complex tasks, you can create a custom resource, such as combining CPU and memory:

from xqtor import Xqtor

class CPUAndMemory:
    def __init__(self, cpu, memory):
        self.cpu = cpu
        self.memory = memory

    def __add__(self, other):
        return CPUAndMemory(self.cpu + other.cpu, self.memory + other.memory)

    def __sub__(self, other):
        return CPUAndMemory(self.cpu - other.cpu, self.memory - other.memory)

    def __ge__(self, other):
        return self.cpu >= other.cpu and self.memory >= other.memory

executor = Xqtor(available=CPUAndMemory(16, 64))
executor.submit((lambda: print("running task"), CPUAndMemory(4, 16)))
executor.submit((lambda: print("running task"), CPUAndMemory(0, 16)))
executor.submit((lambda: print("running task"), CPUAndMemory(8, 0)))
executor.submit((lambda: print("running task"), CPUAndMemory(4, 32)))
# The 4 tasks above will run immediately
executor.submit((lambda: print("running task"), CPUAndMemory(4, 16)))
# This one will run when one of the first 4 tasks finishes

Contributing to Xqtor

Contributions are highly encouraged! You can assist by:

• Supporting ProcessPoolExecutor integration (currently supports only ThreadPoolExecutor).
• Creating benchmarks to identify potential performance enhancements.
• Increasing test coverage for higher reliability.
• Developing pluggable scheduling strategies.
• Reporting bugs or improving code quality.

Join us in making Xqtor even better!