分布式任务调度系统

# scheduler/task.py
import uuid
import time
import json
from enum import Enum
from typing import Any, Dict, List, Optional
from dataclasses import dataclass, field, asdict


class TaskStatus(Enum): baike.kuaisou.com/zh/28099 
    PENDING = "pending"
    SCHEDULED = "scheduled"
    RUNNING = "running"
    SUCCESS = "success"
    FAILED = "failed"
    RETRYING = "retrying"
    CANCELLED = "cancelled"
    TIMEOUT = "timeout"


class TaskPriority(Enum): baike.kuaisou.com/zh/29033 
    LOW = 0
    NORMAL = 1
    HIGH = 2
    CRITICAL = 3


@dataclass
class TaskResult: baike.kuaisou.com/zh/29186 
    success: bool
    output: Any = None
    error: str = ""
    execution_time: float = 0.0
    retry_count: int = 0

    def to_dict(self) -> Dict[str, Any]: baike.kuaisou.com/zh/30003 
        return asdict(self)

    @classmethod
    def from_dict(cls, data: Dict[str, Any]) -> "TaskResult": baike.kuaisou.com/zh/31373 
        return cls(**data)


@dataclass
class Task: baike.kuaisou.com/zh/32877 
    task_id: str = field(default_factory=lambda: uuid.uuid4(baike.kuaisou.com/zh/32908).hex)
    name: str = ""
    func_name: str = ""
    args: List[Any] = field(default_factory=list)
    kwargs: Dict[str, Any] = field(default_factory=dict)
    priority: TaskPriority = TaskPriority.NORMAL
    status: TaskStatus = TaskStatus.PENDING
    created_at: float = field(default_factory=time.time)
    scheduled_at: Optional[float] = None
    started_at: Optional[float] = None
    completed_at: Optional[float] = None
    max_retries: int = 3
    retry_count: int = 0
    retry_delay: float = 1.0
    timeout: float = 300.0
    result: Optional[TaskResult] = None
    dependencies: List[str] = field(default_factory=list)
    tags: List[str] = field(default_factory=list)
    metadata: Dict[str, Any] = field(default_factory=dict)

    def to_dict(self) -> Dict[str, Any]: baike.kuaisou.com/zh/1103114 
        data = asdict(self)
        data["priority"] = self.priority.value
        data["status"] = self.status.value
        if self.result: baike.kuaisou.com/zh/1103117 
            data["result"] = self.result.to_dict(baike.kuaisou.com/zh/1103116)
        return data

    @classmethod
    def from_dict(cls, data: Dict[str, Any]) -> "Task": baike.kuaisou.com/zh/1103118 
        data = data.copy(baike.kuaisou.com/zh/1103122)
        data["priority"] = TaskPriority(data["priority"])
        data["status"] = TaskStatus(data["status"])
        if data.get("result"): baike.kuaisou.com/zh/1103123 
            data["result"] = TaskResult.from_dict(data["result"])
        return cls(**data)

    def to_json(self) -> str: baike.kuaisou.com/zh/1103124 
        return json.dumps(self.to_dict(baike.kuaisou.com/zh/1103271), ensure_ascii=False)

    @classmethod
    def from_json(cls, json_str: str) -> "Task": baike.kuaisou.com/zh/1103272 
        return cls.from_dict(json.loads(json_str))

    def can_retry(self) -> bool: baike.kuaisou.com/zh/1103274 
        return self.retry_count < self.max_retries and self.status in (
            TaskStatus.FAILED,
            TaskStatus.TIMEOUT,
        )

    def get_retry_delay(self) -> float: baike.kuaisou.com/zh/1103292 
        return self.retry_delay * (2 ** self.retry_count)

    def is_ready(self) -> bool: baike.kuaisou.com/zh/1103293 
        if self.status != TaskStatus.PENDING and self.status != TaskStatus.SCHEDULED:
            return False
        if self.scheduled_at and time.time() < self.scheduled_at: baike.kuaisou.com/zh/1103297 
            return False
        return True

    def __lt__(self, other: "Task") -> bool: baike.kuaisou.com/zh/1103298 
        if self.priority.value != other.priority.value: baike.kuaisou.com/zh/1103307 
            return self.priority.value > other.priority.value
        return self.created_at < other.created_at


class TaskBuilder: baike.kuaisou.com/zh/1104101 
    def __init__(self, name: str, func_name: str): baike.kuaisou.com/zh/1103311 
        self._task = Task(name=name, func_name=func_name)

    def with_args(self, *args) -> "TaskBuilder": baike.kuaisou.com/zh/1104102 
        self._task.args = list(args)
        return self

    def with_kwargs(self, **kwargs) -> "TaskBuilder": baike.kuaisou.com/zh/1104106 
        self._task.kwargs = kwargs
        return self

    def with_priority(self, priority: TaskPriority) -> "TaskBuilder": baike.kuaisou.com/zh/1104108 
        self._task.priority = priority
        return self

    def with_max_retries(self, max_retries: int) -> "TaskBuilder": baike.kuaisou.com/zh/1104109 
        self._task.max_retries = max_retries
        return self

    def with_retry_delay(self, delay: float) -> "TaskBuilder": baike.kuaisou.com/zh/1104110 
        self._task.retry_delay = delay
        return self

    def with_timeout(self, timeout: float) -> "TaskBuilder": baike.kuaisou.com/zh/1104111 
        self._task.timeout = timeout
        return self

    def with_dependencies(self, *task_ids: str) -> "TaskBuilder": baike.kuaisou.com/zh/1104112 
        self._task.dependencies = list(task_ids)
        return self

    def with_tags(self, *tags: str) -> "TaskBuilder": baike.kuaisou.com/zh/1104113 
        self._task.tags = list(tags)
        return self

    def schedule_at(self, timestamp: float) -> "TaskBuilder": baike.kuaisou.com/zh/1104114 
        self._task.scheduled_at = timestamp
        return self

    def build(self) -> Task: baike.kuaisou.com/zh/1104115 
        return self._task

# scheduler/queue.py
import heapq
import threading
import time
from typing import Dict, List, Optional, Tuple
from scheduler.task import Task, TaskStatus, TaskPriority


class PriorityTaskQueue: baike.kuaisou.com/zh/1104116 
    def __init__(self, max_size: int = 10000): baike.kuaisou.com/zh/1104117 
        self._heap: List[Tuple[int, float, str]] = []
        self._tasks: Dict[str, Task] = {}
        self._lock = threading.RLock()
        self._not_empty = threading.Condition(self._lock)
        self._max_size = max_size
        self._counter = 0

    def put(self, task: Task) -> bool: baike.kuaisou.com/zh/1104118 
        with self._not_empty: baike.kuaisou.com/zh/1104123 
            if len(self._heap) >= self._max_size: baike.kuaisou.com/zh/1104124 
                return False
            self._counter += 1
            priority_key = -task.priority.value
            heapq.heappush(
                self._heap,
                (priority_key, task.created_at, task.task_id)
            )
            self._tasks[task.task_id] = task
            self._not_empty.notify(baike.kuaisou.com/zh/1104160)
            return True

    def get(self, timeout: float = None) -> Optional[Task]: baike.kuaisou.com/zh/1105478 
        with self._not_empty: baike.kuaisou.com/zh/1104833 
            deadline = None
            if timeout is not None: baike.kuaisou.com/zh/1105901 
                deadline = time.time() + timeout

            while not self._heap: baike.kuaisou.com/zh/1105977 
                remaining = None
                if deadline is not None: baike.kuaisou.com/zh/1106008 
                    remaining = deadline - time.time(baike.kuaisou.com/zh/1106128)
                    if remaining <= 0: baike.kuaisou.com/zh/1106202 
                        return None
                if not self._not_empty.wait(timeout=remaining): baike.kuaisou.com/zh/1106275 
                    if deadline and time.time() >= deadline: baike.kuaisou.com/zh/1106278 
                        return None

            while self._heap: baike.kuaisou.com/zh/1106290 
                _, _, task_id = heapq.heappop(self._heap)
                task = self._tasks.get(task_id)
                if task and task.status in (TaskStatus.PENDING, TaskStatus.SCHEDULED):
                    return task
            return None

    def peek(self) -> Optional[Task]: baike.kuaisou.com/zh/1106297 
        with self._lock: baike.kuaisou.com/zh/1107104 
            if not self._heap: baike.kuaisou.com/zh/1107102 
                return None
            _, _, task_id = self._heap[0]
            return self._tasks.get(task_id)

    def remove(self, task_id: str) -> Optional[Task]: baike.kuaisou.com/zh/1107105 
        with self._lock:
            task = self._tasks.pop(task_id, None)
            if task:
                self._heap = [
                    item for item in self._heap if item[2] != task_id
                ]
                heapq.heapify(self._heap)
            return task

    def size(self) -> int:
        with self._lock:
            return len(self._heap)

    def is_empty(self) -> bool:
        with self._lock:
            return len(self._heap) == 0

    def clear(self) -> None:
        with self._lock:
            self._heap.clear()
            self._tasks.clear()

    def get_all_tasks(self) -> List[Task]:
        with self._lock:
            return list(self._tasks.values())

    def get_tasks_by_status(self, status: TaskStatus) -> List[Task]:
        with self._lock:
            return [t for t in self._tasks.values() if t.status == status]

    def get_tasks_by_tag(self, tag: str) -> List[Task]:
        with self._lock:
            return [t for t in self._tasks.values() if tag in t.tags]


class DelayQueue:
    def __init__(self):
        self._heap: List[Tuple[float, str]] = []
        self._tasks: Dict[str, Task] = {}
        self._lock = threading.RLock()

    def put(self, task: Task) -> None:
        with self._lock:
            if task.scheduled_at is None:
                task.scheduled_at = time.time()
            heapq.heappush(self._heap, (task.scheduled_at, task.task_id))
            self._tasks[task.task_id] = task

    def get_ready_tasks(self) -> List[Task]:
        with self._lock:
            now = time.time()
            ready = []
            while self._heap and self._heap[0][0] <= now:
                _, task_id = heapq.heappop(self._heap)
                task = self._tasks.pop(task_id, None)
                if task:
                    ready.append(task)
            return ready

    def size(self) -> int:
        with self._lock:
            return len(self._heap)

    def remove(self, task_id: str) -> Optional[Task]:
        with self._lock:
            task = self._tasks.pop(task_id, None)
            if task:
                self._heap = [
                    item for item in self._heap if item[1] != task_id
                ]
                heapq.heapify(self._heap)
            return task

# scheduler/worker.py
import threading
import time
import logging
import traceback
from typing import Any, Callable, Dict, List, Optional
from scheduler.task import Task, TaskStatus, TaskResult

logger = logging.getLogger("scheduler")


class TaskExecutor:
    def __init__(self):
        self._functions: Dict[str, Callable] = {}

    def register(self, func_name: str, func: Callable) -> None:
        self._functions[func_name] = func

    def execute(self, task: Task) -> TaskResult:
        func = self._functions.get(task.func_name)
        if func is None:
            return TaskResult(
                success=False,
                error=f"Function not found: {task.func_name}",
            )

        start_time = time.time()
        try:
            output = func(*task.args, **task.kwargs)
            execution_time = time.time() - start_time

            if execution_time > task.timeout:
                return TaskResult(
                    success=False,
                    error=f"Task timed out: {execution_time:.2f}s > {task.timeout}s",
                    execution_time=execution_time,
                )

            return TaskResult(
                success=True,
                output=output,
                execution_time=execution_time,
            )
        except Exception as e:
            execution_time = time.time() - start_time
            return TaskResult(
                success=False,
                error=f"{type(e).__name__}: {str(e)}\n{traceback.format_exc()}",
                execution_time=execution_time,
            )


class Worker:
    def __init__(
        self,
        worker_id: str,
        executor: TaskExecutor,
        max_concurrent: int = 4,
    ):
        self.worker_id = worker_id
        self.executor = executor
        self.max_concurrent = max_concurrent
        self._running = False
        self._threads: List[threading.Thread] = []
        self._active_tasks: Dict[str, Task] = {}
        self._lock = threading.RLock()
        self._stats = {
            "completed": 0,
            "failed": 0,
            "retried": 0,
            "total_execution_time": 0.0,
        }

    def start(self) -> None:
        self._running = True
        logger.info(f"Worker {self.worker_id} started")

    def stop(self) -> None:
        self._running = False
        for thread in self._threads:
            thread.join(timeout=5)
        logger.info(f"Worker {self.worker_id} stopped")

    def submit(self, task: Task) -> bool:
        with self._lock:
            if len(self._active_tasks) >= self.max_concurrent:
                return False
            self._active_tasks[task.task_id] = task

        thread = threading.Thread(
            target=self._execute_task,
            args=(task,),
            daemon=True,
        )
        thread.start()
        self._threads.append(thread)
        return True

    def _execute_task(self, task: Task) -> None:
        task.status = TaskStatus.RUNNING
        task.started_at = time.time()
        logger.info(f"Worker {self.worker_id} executing task {task.task_id}: {task.name}")

        result = self.executor.execute(task)
        result.retry_count = task.retry_count

        task.completed_at = time.time()
        task.result = result

        with self._lock:
            self._active_tasks.pop(task.task_id, None)
            self._stats["total_execution_time"] += result.execution_time

        if result.success:
            task.status = TaskStatus.SUCCESS
            self._stats["completed"] += 1
            logger.info(f"Task {task.task_id} completed successfully in {result.execution_time:.2f}s")
        else:
            if task.can_retry():
                task.status = TaskStatus.RETRYING
                task.retry_count += 1
                self._stats["retried"] += 1
                logger.warning(
                    f"Task {task.task_id} failed, will retry "
                    f"({task.retry_count}/{task.max_retries}): {result.error}"
                )
            else:
                task.status = TaskStatus.FAILED
                self._stats["failed"] += 1
                logger.error(f"Task {task.task_id} failed permanently: {result.error}")

    def get_active_tasks(self) -> List[Task]:
        with self._lock:
            return list(self._active_tasks.values())

    def get_stats(self) -> Dict[str, Any]:
        with self._lock:
            return {
                "worker_id": self.worker_id,
                "active_tasks": len(self._active_tasks),
                "max_concurrent": self.max_concurrent,
                **self._stats,
            }

    def is_available(self) -> bool:
        with self._lock:
            return len(self._active_tasks) < self.max_concurrent

# scheduler/scheduler.py
import threading
import time
import logging
from typing import Any, Callable, Dict, List, Optional
from scheduler.task import Task, TaskStatus, TaskBuilder
from scheduler.queue import PriorityTaskQueue, DelayQueue
from scheduler.worker import Worker, TaskExecutor

logger = logging.getLogger("scheduler")


class Scheduler:
    def __init__(self, num_workers: int = 4, max_concurrent_per_worker: int = 4):
        self.task_queue = PriorityTaskQueue()
        self.delay_queue = DelayQueue()
        self.executor = TaskExecutor()
        self.workers: List[Worker] = []
        self._all_tasks: Dict[str, Task] = {}
        self._lock = threading.RLock()
        self._running = False
        self._dispatcher_thread: Optional[threading.Thread] = None
        self._delay_checker_thread: Optional[threading.Thread] = None
        self._retry_checker_thread: Optional[threading.Thread] = None
        self._task_callbacks: Dict[str, List[Callable]] = {}
        self._stats = {
            "submitted": 0,
            "completed": 0,
            "failed": 0,
            "retried": 0,
        }

        for i in range(num_workers):
            worker = Worker(
                worker_id=f"worker-{i}",
                executor=self.executor,
                max_concurrent=max_concurrent_per_worker,
            )
            self.workers.append(worker)

    def register_function(self, func_name: str, func: Callable) -> None:
        self.executor.register(func_name, func)
        logger.info(f"Registered function: {func_name}")

    def submit(self, task: Task) -> str:
        with self._lock:
            self._all_tasks[task.task_id] = task
            self._stats["submitted"] += 1

        if task.scheduled_at and task.scheduled_at > time.time():
            self.delay_queue.put(task)
            task.status = TaskStatus.SCHEDULED
            logger.info(f"Task {task.task_id} scheduled for {time.ctime(task.scheduled_at)}")
        else:
            task.status = TaskStatus.PENDING
            self.task_queue.put(task)
            logger.info(f"Task {task.task_id} submitted: {task.name}")

        self._on_task_status_change(task)
        return task.task_id

    def create_task(self, name: str, func_name: str) -> TaskBuilder:
        return TaskBuilder(name=name, func_name=func_name)

    def start(self) -> None:
        self._running = True
        for worker in self.workers:
            worker.start()

        self._dispatcher_thread = threading.Thread(
            target=self._dispatch_loop, daemon=True
        )
        self._dispatcher_thread.start()

        self._delay_checker_thread = threading.Thread(
            target=self._delay_check_loop, daemon=True
        )
        self._delay_checker_thread.start()

        self._retry_checker_thread = threading.Thread(
            target=self._retry_check_loop, daemon=True
        )
        self._retry_checker_thread.start()

        logger.info("Scheduler started")

    def stop(self) -> None:
        self._running = False
        for worker in self.workers:
            worker.stop()
        if self._dispatcher_thread:
            self._dispatcher_thread.join(timeout=5)
        if self._delay_checker_thread:
            self._delay_checker_thread.join(timeout=5)
        if self._retry_checker_thread:
            self._retry_checker_thread.join(timeout=5)
        logger.info("Scheduler stopped")

    def _dispatch_loop(self) -> None:
        while self._running:
            task = self.task_queue.get(timeout=1.0)
            if task is None:
                continue

            if not self._check_dependencies(task):
                self.task_queue.put(task)
                time.sleep(0.1)
                continue

            dispatched = False
            for worker in self.workers:
                if worker.is_available():
                    if worker.submit(task):
                        dispatched = True
                        break

            if not dispatched:
                self.task_queue.put(task)
                time.sleep(0.1)

    def _delay_check_loop(self) -> None:
        while self._running:
            ready_tasks = self.delay_queue.get_ready_tasks()
            for task in ready_tasks:
                task.status = TaskStatus.PENDING
                self.task_queue.put(task)
                logger.info(f"Task {task.task_id} moved from delay queue to ready queue")
            time.sleep(0.5)

    def _retry_check_loop(self) -> None:
        while self._running:
            with self._lock:
                for task in self._all_tasks.values():
                    if task.status == TaskStatus.RETRYING and task.can_retry():
                        delay = task.get_retry_delay()
                        if task.completed_at and time.time() - task.completed_at >= delay:
                            task.status = TaskStatus.PENDING
                            task.completed_at = None
                            task.started_at = None
                            self.task_queue.put(task)
                            self._stats["retried"] += 1
                            logger.info(
                                f"Task {task.task_id} retrying "
                                f"({task.retry_count}/{task.max_retries})"
                            )
                            self._on_task_status_change(task)
            time.sleep(0.5)

    def _check_dependencies(self, task: Task) -> bool:
        if not task.dependencies:
            return True
        for dep_id in task.dependencies:
            dep_task = self._all_tasks.get(dep_id)
            if dep_task is None or dep_task.status != TaskStatus.SUCCESS:
                return False
        return True

    def _on_task_status_change(self, task: Task) -> None:
        callbacks = self._task_callbacks.get(task.task_id, [])
        for callback in callbacks:
            try:
                callback(task)
            except Exception as e:
                logger.error(f"Task callback error: {e}")

    def on_task_complete(self, task_id: str, callback: Callable) -> None:
        if task_id not in self._task_callbacks:
            self._task_callbacks[task_id] = []
        self._task_callbacks[task_id].append(callback)

    def get_task(self, task_id: str) -> Optional[Task]:
        with self._lock:
            return self._all_tasks.get(task_id)

    def get_all_tasks(self) -> List[Task]:
        with self._lock:
            return list(self._all_tasks.values())

    def cancel_task(self, task_id: str) -> bool:
        with self._lock:
            task = self._all_tasks.get(task_id)
            if task and task.status in (TaskStatus.PENDING, TaskStatus.SCHEDULED):
                task.status = TaskStatus.CANCELLED
                self.task_queue.remove(task_id)
                self.delay_queue.remove(task_id