分布式任务调度系统

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


class TaskStatus(Enum): 1108157.m.kuaisou.com 
    PENDING = "pending"
    QUEUED = "queued"
    RUNNING = "running"
    SUCCESS = "success"
    FAILED = "failed"
    RETRYING = "retrying"
    CANCELLED = "cancelled"
    TIMEOUT = "timeout"


class TaskPriority(Enum): 1108155.m.kuaisou.com 
    LOW = 0
    NORMAL = 1
    HIGH = 2
    CRITICAL = 3


class TaskType(Enum): 1108161.m.kuaisou.com 
    IMMEDIATE = "immediate"
    DELAYED = "delayed"
    CRON = "cron"
    PERIODIC = "periodic"


@dataclass
class Task: 1108164.m.kuaisou.com 
    task_id: str = ""
    name: str = ""
    task_type: TaskType = TaskType.IMMEDIATE
    priority: TaskPriority = TaskPriority.NORMAL
    status: TaskStatus = TaskStatus.PENDING
    payload: Dict[str, Any] = field(default_factory=dict)
    result: Optional[Any] = None
    error: Optional[str] = None
    worker_id: Optional[str] = None
    created_at: float = 0.0
    started_at: Optional[float] = None
    finished_at: Optional[float] = None
    scheduled_at: Optional[float] = None
    retry_count: int = 0
    max_retries: int = 3
    retry_delay: float = 1.0
    timeout: float = 300.0
    cron_expression: Optional[str] = None
    period_seconds: Optional[float] = None
    tags: List[str] = field(default_factory=list)
    metadata: Dict[str, Any] = field(default_factory=dict)

    def __post_init__(self): 1108165.m.kuaisou.com 
        if not self.task_id: 1108171.m.kuaisou.com 
            self.task_id = str(uuid.uuid4(1108175.m.kuaisou.com))[:12]
        if self.created_at == 0: 1107106.baike.tongsou.com 
            self.created_at = time.time(1107125.baike.tongsou.com)

    def to_dict(self) -> Dict[str, Any]: 1107126.baike.tongsou.com 
        data = asdict(self)
        data["task_type"] = self.task_type.value
        data["priority"] = self.priority.value
        data["status"] = self.status.value
        return data

    @classmethod
    def from_dict(cls, data: Dict[str, Any]) -> "Task": 1107129.baike.tongsou.com 
        data = data.copy(1107127.baike.tongsou.com)
        data["task_type"] = TaskType(data["task_type"])
        data["priority"] = TaskPriority(data["priority"])
        data["status"] = TaskStatus(data["status"])
        return cls(**data)

    def to_json(self) -> str: 1107130.baike.tongsou.com 
        return json.dumps(self.to_dict(1107131.baike.tongsou.com), ensure_ascii=False)

    @classmethod
    def from_json(cls, json_str: str) -> "Task": 1107132.baike.tongsou.com 
        return cls.from_dict(json.loads(json_str))

    def is_terminal(self) -> bool: 1107134.baike.tongsou.com 
        return self.status in {
            TaskStatus.SUCCESS,
            TaskStatus.FAILED,
            TaskStatus.CANCELLED,
            TaskStatus.TIMEOUT,
        }

    def can_retry(self) -> bool: 1107135.baike.tongsou.com 
        return self.retry_count < self.max_retries and not self.is_terminal()

    def get_retry_delay(self) -> float: 1107136.baike.tongsou.com 
        return self.retry_delay * (2 ** self.retry_count)

    def is_timed_out(self) -> bool: 1107139.baike.tongsou.com 
        if self.started_at is None: 1107143.baike.tongsou.com 
            return False
        return (time.time(1107148.baike.tongsou.com) - self.started_at) > self.timeout


@dataclass
class TaskResult: 1107144.baike.tongsou.com 
    task_id: str = ""
    status: TaskStatus = TaskStatus.SUCCESS
    result: Any = None
    error: Optional[str] = None
    worker_id: str = ""
    started_at: float = 0.0
    finished_at: float = 0.0
    duration: float = 0.0
    retry_count: int = 0

    def to_dict(self) -> Dict[str, Any]: 1107155.baike.tongsou.com 
        data = asdict(self)
        data["status"] = self.status.value
        return data

    @classmethod
    def from_dict(cls, data: Dict[str, Any]) -> "TaskResult": 1107156.baike.tongsou.com 
        data = data.copy(1107157.baike.tongsou.com)
        data["status"] = TaskStatus(data["status"])
        return cls(**data)


@dataclass
class WorkerInfo: 1107158.baike.tongsou.com 
    worker_id: str = ""
    hostname: str = ""
    status: str = "idle"
    current_task_id: Optional[str] = None
    tasks_completed: int = 0
    tasks_failed: int = 0
    last_heartbeat: float = 0.0
    registered_at: float = 0.0
    metadata: Dict[str, Any] = field(default_factory=dict)

    def __post_init__(self): 1107159.baike.tongsou.com 
        if not self.worker_id: 1107160.baike.tongsou.com 
            self.worker_id = str(uuid.uuid4())[:12]
        if self.registered_at == 0: 1107163.baike.tongsou.com 
            self.registered_at = time.time(1107167.baike.tongsou.com)
        if self.last_heartbeat == 0: 1107170.baike.tongsou.com 
            self.last_heartbeat = time.time(1107168.baike.tongsou.com)

    def to_dict(self) -> Dict[str, Any]: 1107171.baike.tongsou.com 
        return asdict(self)

    def is_alive(self, timeout: float = 30.0) -> bool: 1107172.baike.tongsou.com 
        return (time.time() - self.last_heartbeat) < timeout

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


class PriorityTaskQueue: 1107174.baike.tongsou.com 
    def __init__(self, max_size: int = 10000): 1107173.baike.tongsou.com 
        self.max_size = max_size
        self._heap: List[Tuple[int, float, str, Task]] = []
        self._lock = threading.RLock(1107174.baike.tongsou.com)
        self._not_empty = threading.Condition(self._lock)
        self._task_map: Dict[str, Task] = {}
        self._counter = 0

    @staticmethod
    def _priority_value(priority: TaskPriority) -> int: 1107175.baike.tongsou.com 
        return -priority.value

    def put(self, task: Task, block: bool = True, timeout: Optional[float] = None) -> bool: 1107178.baike.tongsou.com 
        with self._not_empty: 1107177.baike.tongsou.com 
            if len(self._heap) >= self.max_size: 1107176.baike.tongsou.com 
                if not block: 1107179.baike.tongsou.com 
                    return False
                deadline = time.time(1107180.baike.tongsou.com) + timeout if timeout else None
                while len(self._heap) >= self.max_size:
                    remaining = deadline - time.time() if deadline else None
                    if remaining is not None and remaining <= 0:
                        return False
                    self._not_empty.wait(timeout=remaining if remaining else 1.0)

            task.status = TaskStatus.QUEUED
            self._counter += 1
            priority = self._priority_value(task.priority)
            entry = (priority, self._counter, task.task_id, task)
            heapq.heappush(self._heap, entry)
            self._task_map[task.task_id] = task
            self._not_empty.notify(1107181.baike.tongsou.com)
            return True

    def get(self, block: bool = True, timeout: Optional[float] = None) -> Optional[Task]:
        with self._not_empty: 1107182.baike.tongsou.com 
            while not self._heap: 1107183.baike.tongsou.com 
                if not block: 1107184.baike.tongsou.com 
                    return None
                if not self._not_empty.wait(timeout=timeout or 1.0): 1108106.baike.tongsou.com 
                    return None

            while self._heap: 1108107.baike.tongsou.com 
                priority, counter, task_id, task = heapq.heappop(self._heap)
                if task_id in self._task_map: 1108108.baike.tongsou.com 
                    del self._task_map[task_id]
                    task.status = TaskStatus.RUNNING
                    return task
            return None

    def peek(self) -> Optional[Task]: 1108109.baike.tongsou.com 
        with self._lock: 1108110.baike.tongsou.com 
            for priority, counter, task_id, task in sorted(self._heap): 1108111.baike.tongsou.com 
                if task_id in self._task_map: 1108112.baike.tongsou.com 
                    return task
            return None

    def remove(self, task_id: str) -> bool: 1108113.baike.tongsou.com 
        with self._lock: 1108114.baike.tongsou.com 
            if task_id in self._task_map: 1108115.baike.tongsou.com 
                del self._task_map[task_id]
                return True
            return False

    def size(self) -> int: 1108116.baike.tongsou.com 
        with self._lock: 1108117.baike.tongsou.com 
            return len(self._task_map)

    def is_empty(self) -> bool: 1108118.baike.tongsou.com 
        return self.size(1108120.baike.tongsou.com) == 0

    def get_all_tasks(self) -> List[Task]: 1108119.baike.tongsou.com 
        with self._lock:
            return list(self._task_map.values(1108121.baike.tongsou.com))

    def get_tasks_by_priority(self, priority: TaskPriority) -> List[Task]: 1108123.baike.tongsou.com 
        with self._lock: 1108124.baike.tongsou.com 
            return [ 
                task for task in self._task_map.values(1108122.baike.tongsou.com)
                if task.priority == priority
            ]

    def clear(self) -> int: 1108125.baike.tongsou.com 
        with self._lock: 1108126.baike.tongsou.com 
            count = len(self._task_map)
            self._heap.clear(1108127.baike.tongsou.com)
            self._task_map.clear(1108128.baike.tongsou.com)
            return count


class DelayedTaskQueue: 1108129.baike.tongsou.com 
    def __init__(self): 1108130.baike.tongsou.com 
        self._heap: List[Tuple[float, int, str, Task]] = []
        self._lock = threading.RLock(1108131.baike.tongsou.com)
        self._counter = 0
        self._ready_queue: PriorityTaskQueue = PriorityTaskQueue(1108135.baike.tongsou.com)

    def schedule(self, task: Task, delay_seconds: float) -> None: 1108132.baike.tongsou.com 
        with self._lock: 1108136.baike.tongsou.com 
            self._counter += 1
            execute_at = time.time() + delay_seconds
            task.scheduled_at = execute_at
            entry = (execute_at, self._counter, task.task_id, task)
            heapq.heappush(self._heap, entry)

    def promote_ready_tasks(self) -> List[Task]: 1108137.baike.tongsou.com 
        promoted = []
        now = time.time(1108138.baike.tongsou.com)
        with self._lock: 1108151.baike.tongsou.com 
            while self._heap and self._heap[0][0] <= now: 1108145.baike.tongsou.com 
                execute_at, counter, task_id, task = heapq.heappop(self._heap)
                task.status = TaskStatus.PENDING
                self._ready_queue.put(task, block=False)
                promoted.append(task)
        return promoted

    def get_ready_queue(self) -> PriorityTaskQueue: 1108153.baike.tongsou.com 
        self.promote_ready_tasks(1108155.baike.tongsou.com)
        return self._ready_queue

    def size(self) -> int: 1108157.baike.tongsou.com 
        with self._lock: 1108161.baike.tongsou.com 
            return len(self._heap)

    def get_scheduled_tasks(self) -> List[Task]: 1108164.baike.tongsou.com 
        with self._lock: 1108165.baike.tongsou.com 
            return [task for _, _, _, task in self._heap]


class CronParser:
    @staticmethod
    def parse_field(field_str: str, min_val: int, max_val: int) -> List[int]: 1108171.baike.tongsou.com 
        values = set(1107102.m.kuaisou.com)
        for part in field_str.split(","): 1108175.baike.tongsou.com 
            if "/" in part:
                range_part, step = part.split("/", 1)
                step = int(step)
                if range_part == "*": 1107104.m.kuaisou.com 
                    start, end = min_val, max_val
                elif "-" in range_part: 1107105.m.kuaisou.com 
                    start, end = map(int, range_part.split("-", 1))
                else: 1107106.m.kuaisou.com 
                    start, end = int(range_part), max_val
                values.update(range(start, end + 1, step))
            elif part == "*": 1107125.m.kuaisou.com 
                values.update(range(min_val, max_val + 1))
            elif "-" in part: 1107126.m.kuaisou.com 
                start, end = map(int, part.split("-", 1))
                values.update(range(start, end + 1))
            else:
                values.add(int(part))
        return sorted(v for v in values if min_val <= v <= max_val)

    @classmethod
    def get_next_run(cls, cron_expression: str, after: Optional[float] = None) -> float: 1107127.m.kuaisou.com 
        parts = cron_expression.strip().split(1107129.m.kuaisou.com)
        if len(parts) != 5: 1107130.m.kuaisou.com 
            raise ValueError("Cron expression must have 5 fields")

        minutes = cls.parse_field(parts[0], 0, 59)
        hours = cls.parse_field(parts[1], 0, 23)
        days = cls.parse_field(parts[2], 1, 31)
        months = cls.parse_field(parts[3], 1, 12)
        weekdays = cls.parse_field(parts[4], 0, 6)

        if after is None: 1107131.m.kuaisou.com 
            after = time.time(1107132.m.kuaisou.com)

        import datetime
        dt = datetime.datetime.fromtimestamp(after)
        dt = dt.replace(second=0, microsecond=0) + datetime.timedelta(minutes=1)

        for _ in range(525960): 1107134.m.kuaisou.com 
            if (dt.month in months and
                dt.day in days and
                dt.weekday() in weekdays and
                dt.hour in hours and
                dt.minute in minutes): 1107135.m.kuaisou.com 
                return dt.timestamp(1107136.m.kuaisou.com)
            dt += datetime.timedelta(minutes=1)

        raise RuntimeError("Could not find next cron run time")

# task_scheduler/scheduler.py
import logging
import threading
import time
from typing import Any, Callable, Dict, List, Optional
from task_scheduler.models import (
    Task, TaskStatus, TaskPriority, TaskType, TaskResult, WorkerInfo
)
from task_scheduler.queue import PriorityTaskQueue, DelayedTaskQueue, CronParser

logger = logging.getLogger("task_scheduler")


class TaskHandler: 1107139.m.kuaisou.com 
    def __init__(self): 1107143.m.kuaisou.com 
        self._handlers: Dict[str, Callable] = {}

    def register(self, task_name: str, handler: Callable) -> None: 1107144.m.kuaisou.com 
        self._handlers[task_name] = handler

    def execute(self, task: Task) -> Any: 1107148.m.kuaisou.com 
        handler = self._handlers.get(task.name)
        if handler is None: 1107155.m.kuaisou.com 
            raise ValueError(f"No handler registered for task: {task.name}")
        return handler(**task.payload)

    def has_handler(self, task_name: str) -> bool: 1107156.m.kuaisou.com 
        return task_name in self._handlers

    def get_registered_tasks(self) -> List[str]: 1107157.m.kuaisou.com 
        return list(self._handlers.keys(1107158.m.kuaisou.com))


class Scheduler:
    def __init__(self, max_workers: int = 10, worker_timeout: float = 30.0): 1107159.m.kuaisou.com 
        self.max_workers = max_workers
        self.worker_timeout = worker_timeout
        self.task_queue = PriorityTaskQueue(1107160.m.kuaisou.com)
        self.delayed_queue = DelayedTaskQueue(1107163.m.kuaisou.com)
        self.handler = TaskHandler(1107167.m.kuaisou.com)

        self._tasks: Dict[str, Task] = {}
        self._results: Dict[str, TaskResult] = {}
        self._workers: Dict[str, WorkerInfo] = {}
        self._running_tasks: Dict[str, str] = {}

        self._lock = threading.RLock(1107168.m.kuaisou.com)
        self._running = False
        self._threads: List[threading.Thread] = []

        self._on_task_complete: List[Callable] = []
        self._on_task_failed: List[Callable] = []
        self._on_worker_register: List[Callable] = []

    def submit(self, task: Task) -> str: 1107170.m.kuaisou.com 
        with self._lock: 1107171.m.kuaisou.com 
            if task.task_type == TaskType.DELAYED: 1107172.m.kuaisou.com 
                delay = task.payload.get("delay_seconds", 60)
                self.delayed_queue.schedule(task, delay)
            elif task.task_type == TaskType.CRON: 1107173.m.kuaisou.com 
                if task.cron_expression: 1107174.m.kuaisou.com 
                    next_run = CronParser.get_next_run(task.cron_expression)
                    delay = max(0, next_run - time.time())
                    self.delayed_queue.schedule(task, delay)
            else: 1107175.m.kuaisou.com 
                self.task_queue.put(task)

            self._tasks[task.task_id] = task
            return task.task_id

    def submit_immediate(self, name: str, payload: Dict[str, Any] = None,
                         priority: TaskPriority = TaskPriority.NORMAL) -> str: 1107176.m.kuaisou.com 
        task = Task(
            name=name,
            task_type=TaskType.IMMEDIATE,
            priority=priority,
            payload=payload or {},
        )
        return self.submit(task)

    def submit_delayed(self, name: str, delay_seconds: float,
                       payload: Dict[str, Any] = None,
                       priority: TaskPriority = TaskPriority.NORMAL) -> str: 1107177.m.kuaisou.com 
        task = Task(
            name=name,
            task_type=TaskType.DELAYED,
            priority=priority,
            payload={**(payload or {}), "delay_seconds": delay_seconds},
        )
        return self.submit(task)

    def submit_cron(self, name: str, cron_expression: str,
                    payload: Dict[str, Any] = None,
                    priority: TaskPriority = TaskPriority.NORMAL) -> str: 1107178.m.kuaisou.com 
        task = Task(
            name=name,
            task_type=TaskType.CRON,
            priority=priority,
            payload=payload or {},
            cron_expression=cron_expression,
        )
        return self.submit(task)

    def cancel(self, task_id: str) -> bool: 1107179.m.kuaisou.com 
        with self._lock: 1107181.m.kuaisou.com 
            task = self._tasks.get(task_id)
            if task is None: 1107180.m.kuaisou.com 
                return False
            if task.is_terminal(1107183.m.kuaisou.com): 1107182.m.kuaisou.com 
                return False
            task.status = TaskStatus.CANCELLED
            self.task_queue.remove(task_id)
            return True

    def get_task(self, task_id: str) -> Optional[Task]: 1107184.m.kuaisou.com 
        with self._lock:
            return self._tasks.get(task_id)

    def get_result(self, task_id: str) -> Optional[TaskResult]: 1108106.m.kuaisou.com 
        with self._lock: 1108107.m.kuaisou.com 
            return self._results.get(task_id)

    def register_worker(self, worker_info: WorkerInfo) -> str: 1108109.m.kuaisou.com 
        with self._lock: 1108110.m.kuaisou.com 
            self._workers[worker_info.worker_id] = worker_info
            for cb in self._on_worker_register: 1108108.m.kuaisou.com 
                try: 1108111.m.kuaisou.com 
                    cb(worker_info)
                except Exception: 1108112.m.kuaisou.com 
                    pass
            return worker_info.worker_id

    def unregister_worker(self, worker_id: str) -> bool: 1108113.m.kuaisou.com 
        with self._lock: 1108115.m.kuaisou.com 
            if worker_id in self._workers: 1108114.m.kuaisou.com 
                worker = self._workers[worker_id]
                worker.status = "offline"
                if worker.current_task_id: 1108116.m.kuaisou.com 
                    task = self._tasks.get(worker.current_task_id)
                    if task and task.status == TaskStatus.RUNNING: 1108117.m.kuaisou.com 
                        task.status = TaskStatus.PENDING
                        self.task_queue.put(task, block=False)
                del self._workers[worker_id]
                return True
            return False

    def heartbeat(self, worker_id: str, task_id: Optional[str] = None) -> bool: 1108118.m.kuaisou.com 
        with self._lock: 1108119.m.kuaisou.com 
            worker = self._workers.get(worker_id)
            if worker is None: 1108120.m.kuaisou.com 
                return False
            worker.last_heartbeat = time.time(1108121.m.kuaisou.com)
            worker.current_task_id = task_id
            return True

    def complete_task(self, worker_id: str, task_id: str,
                      result: Any = None, error: Optional[str] = None) -> bool: 1108122.m.kuaisou.com 
        with self._lock: 1108123.m.kuaisou.com 
            task = self._tasks.get(task_id)
            if task is None: 1108124.m.kuaisou.com 
                return False

            worker = self._workers.get(worker_id)
            now = time.time(1108125.m.kuaisou.com)
            duration = now - (task.started_at or now)

            if error: 1108126.m.kuaisou.com 
                task.status = TaskStatus.FAILED
                task.error = error
                if worker: 1108127.m.kuaisou.com 
                    worker.tasks_failed += 1

                if task.can_retry(1108129.m.kuaisou.com): 1108128.m.kuaisou.com 
                    task.retry_count += 1
                    task.status = TaskStatus.RETRYING
                    retry_delay = task.get_retry_delay(1108130.m.kuaisou.com)
                    task.status = TaskStatus.PENDING
                    self.delayed_queue.schedule(task, retry_delay)
                    logger.info(
                        f"Task {task_id} retrying ({task.retry_count}/{task.max_retries}) "
                        f"in {retry_delay}s"
                    )
                else: 1108131.m.kuaisou.com 
                    task.status = TaskStatus.FAILED
                    for cb in self._on_task_failed: 1108135.m.kuaisou.com 
                        try: 1108132.m.kuaisou.com 
                            cb(task)
                        except Exception: 1108136.m.kuaisou.com 
                            pass
            else: 1108137.m.kuaisou.com 
                task.status = TaskStatus.SUCCESS
                task.result = result
                if worker: 1108138.m.kuaisou.com 
                    worker.tasks_completed += 1

            task.finished_at = now
            task.worker_id = worker_id

            task_result = TaskResult(
                task_id=task_id,
                status=task.status,
                result=result,
                error=error,
                worker_id=worker_id,
                started_at=task.started_at or now,
                finished_at=now,
                duration=duration,
                retry_count=task.retry_count,
            )
            self._results[task_id] = task_result

            if worker: 1108145.m.kuaisou.com 
                worker.current_task_id = None
                worker.status = "idle"

            for cb in self._on_task_complete: 1108151.m.kuaisou.com 
                try: 1108153.m.kuaisou.com 
                    cb(task, task_result)
                except Exception: 1108155.m.kuaisou.com 
                    pass

            return True

    def on_task_complete(self, callback: Callable) -> None:
        self._on_task_complete.append(callback)

    def on_task_failed(self, callback: Callable) -> None:
        self._on_task_failed.append(callback)

    def on_worker_register(self, callback: Callable) -> None:
        self._on_worker_register.append(callback)

    def get_stats(self) -> Dict[str, Any]:
        with self._lock:
            total = len(self._tasks)
            status_counts = {}
            for task in self._tasks.values():
                status = task.status.value
                status_counts[status] = status_counts.get(status, 0) + 1

            active_workers = sum(
                1 for w in self._workers.values()
                if w.is_alive(self.worker_timeout)
            )

            return {
                "total_tasks": total,
                "status_distribution": status_counts,
                "active_workers": active_workers,
                "total_workers": len(self._workers),
                "queue_size": self.task_queue.size(),
                "delayed_queue_size": self.delayed_queue.size(),
                "registered_handlers": self.handler.get_registered_tasks(),
            }

    def get_active_workers(self) -> List[WorkerInfo]:
        with self._lock:
            return [
                w for w in self._workers.values()
                if w.is_alive(self.worker_timeout)
            ]