def ConfigManager(config, need_reload=False, defaults=None):
    return _configmanager.get_config(config, need_reload, defaults) class Task(AbstractTask):
    def __init__(self, name, scheduler_ref=None, **taskinfo):
        super(Task, self).__init__(name, scheduler_ref, **taskinfo)

        try:
            self.upstream = taskinfo['upstream[]']
            self.rsyncdir = taskinfo['rsyncdir']
            if self.rsyncdir[-1] != '/':
                self.rsyncdir += '/'
            self.localdir = taskinfo['localdir']
            if not os.path.exists(self.localdir):
                try:
                    os.makedirs(self.localdir, 0o755)
                except:
                    log.error("Error when create directory: %s", self.localdir)
        except KeyError as e:
            log.error("Error in config for mirror: %s, key: %s not found.", self.name, e)
            self.enabled  = False

        self.exclude  = taskinfo['exclude'] if "exclude" in taskinfo else None
        self.args     = taskinfo['args']    if "args" in taskinfo    else DEFAULT_ARGS

    def get_args(self, stage = 1):
        args  = [os.path.basename(self.command)]
        args += shlex.split(self.args)
        if self.twostage and stage == 2:
            args.remove("--delete")
            args.append("--delete-after")
        if self.exclude:
            args += shlex.split(self.exclude)
        if self.twostage and stage == 1:
            args += [self.upstream[0] + '::' + self.rsyncdir + self.firststage + '/',\
                     self.localdir + '/' + self.firststage]
            return args
        if self.twostage and stage == 2:
            args += ['--exclude', '/' + self.firststage + '/']
        args += [self.upstream[0] + '::' + self.rsyncdir,\
                 self.localdir]
        return args class SimpleTask(AbstractTask):
    def __init__(self, name, scheduler_ref=None, **taskinfo):
        super(SimpleTask, self).__init__(name, scheduler_ref, **taskinfo)

        """
        If a SimpleTask is twostage, args is interpreted as args for second
        stage, firststage is interpreted as args for first stage.

        """
        self.args = taskinfo['args'] if "args" in taskinfo else None

    def get_args(self, stage = 1):
        args  = [os.path.basename(self.command)]
        if self.twostage and stage == 1:
            args += self.firststage.split(" ")
            return args
        if self.args:
            args += shlex.split(self.args)
        return args TASK_TYPES = { "simple" : SimpleTask } PRIORITY_MIN = 1  # high priority PRIORITY_MAX = 10 # low  priority REGULAR_TASK = 1  # regular task TIMEOUT_TASK = 2  # timeout checking task SYSTEM_TASK  = 3  # system internal task, run in thread, so this is different def loadavg(duration = 5):
    if duration not in d:
        duration = 5
    if hasattr(os, "getloadavg"):
        return os.getloadavg()[d[duration]]
    try:
        loadavgs = open("/proc/loadavg").read().split(" ")
        return float(loadavgs[d[duration]])
    except:
        return 0.0 def tcpconn(port = 80):
    try:
        # TODO: using AF_NETLINK to fetch tcp information instead from /proc
        sock = socket.socket(socket.AF_NETLINK, socket.SOCK_RAW)
        sock.close()
    except Exception as e:
        pass
    connections = 0
    tcp   = re.compile(pattern)
    files = ("/proc/net/tcp", "/proc/net/tcp6")
    for path in files:
        if os.access(path, os.R_OK):
            fp = open(path)
            fp.readline() # skip title
            for line in fp:
                conn = tcp.search(line).groupdict()
                local_port = int(conn['local_port'], 16)
                if local_port != port:
                    continue
                status = int(conn['status'], 16)
                if status == TCP_STATUS.TCP_ESTABLISHED:
                    connections += 1
            fp.close()
    return connections class TaskInfo():
    def __init__(self, name, tasktype, time, priority):
        self.name     = name
        self.tasktype = tasktype
        self.time     = time
        self.priority = priority

    def __eq__(self, other):
        return self.name == other.name

    def __lt__(self, other):
        return (self.time - other.time < 0) if self.time != other.time else (
                                           self.priority - other.priority < 0)

    def __cmp__(self, other):
        return (self.time - other.time) if self.time != other.time else (
                                           self.priority - other.priority)

    def __str__(self):
        return "(name: %s, type: %d, time: %d, priority: %d)" % (self.name,
                                    self.tasktype, self.time, self.priority) class Queue(object):
    def __init__(self, *items):
        self.queue = []
        for item in items:
            if not isinstance(item, TaskInfo):
                continue
            self.put(item)

    def put(self, item):
        bisect.insort(self.queue, item)

    def get(self):
        try:
            return self.queue.pop(0)
        except:
            return None

    def remove(self, item):
        self.queue.remove(item)

    def empty(self):
        return ( len(self.queue) == 0 )

    def size(self, key = None, value = None):
        if key is None or value is None:
            return len(self.queue)
        else:
            if self.empty():
                return 0
            if not hasattr(self.queue[0], key):
                return 0
            count = 0
            for item in self.queue:
                count += ( getattr(item, key) == value )
            return count

    def find(self, name):
        for taskinfo in self.queue:
            if taskinfo.name == name:
                return taskinfo
        return None

    def __getitem__(self, key):
        try:
            return self.queue[key]
        except:
            return None

    def __str__(self):
        return "[" + ",".join([str(item) for item in self.queue]) + "]"

    def __iter__(self):
        for item in self.queue:
            yield item

    def __contains__(self, item):
        return item in self.queue class Component(object):

    def __init__(self, name, depend = None):
        self._name   = name
        self._depend = depend
        _component_registry.register(self)

    def start(self):
        pass

    def stop(self):
        pass
import os, sys
import time
import fcntl
import signal
import logging
import weakref
import mmap
import struct
    import cPickle as pickle
    import pickle
import mirror.common
import mirror.error
import mirror.handler
import mirror.event
import mirror.component   as     component
        import re
from mirror.configmanager import ConfigManager
from mirror.task          import Task, SimpleTask, TASK_TYPES
from mirror.task          import PRIORITY_MIN, PRIORITY_MAX
from mirror.task          import REGULAR_TASK, TIMEOUT_TASK, SYSTEM_TASK
from mirror.sysinfo       import loadavg, tcpconn
from mirror.queue         import TaskInfo, Queue
from mirror.component     import Component
from collections import OrderedDict as odict
        """
        Stop a task, it should only be called when that task timeouts.

        """
        if taskinfo.name not in self.tasks:
            return
        task = self.tasks[taskinfo.name]
        if not task.running:
            return
        pid  = task.pid
        # Python's SIGCHLD sometimes has delay in calling its handler,
        # we have to disable sigchld_handler() here.
        # More: http://utcc.utoronto.ca/~cks/space/blog/python/CPythonSignals
        signal.signal(signal.SIGCHLD, signal.SIG_DFL)
        task.stop()
        self.stop_task_manually(task, pid)
        signal.signal(signal.SIGCHLD, mirror.handler.sigchld_handler)

    def stop_task_manually(self, task, pid):
        """
        Without SIGCHLD handler, we have to waitpid() here.

        """
        pid, status  = os.waitpid(pid, 0)
        endstr, code = self.parse_return_status(status)
        task.code    = code
        log.info("Killed task: %s %s %d, pid %d", task.name, endstr, code, pid)
        self.remove_timeout_task(task.name)
        self.task_finished(task)

    def stop_task_with_pid(self, pid, status):
        """
        This is called when we got a SIGCHLD signal.
        Change task's running and pid attr as it's stopped.

        """
        self.roused_by_child = True
        for taskname, task in self.tasks.items():
            if task.isinternal:
                continue
            if task.pid == pid:
                if not task.running:
                    return
                endstr, code = self.parse_return_status(status)
                task.code    = code
                log.info("Task: %s %s %d, pid %d", taskname, endstr, code, pid)
                self.remove_timeout_task(taskname)
                self.task_finished(task)
                return

    def task_finished(self, task):
        """
        Check whether a task needs post process, e.g. two stage tasks.

        """
        event_manager = component.get("EventManager")
        if not task.twostage:
            event_manager.emit(mirror.event.TaskStopEvent(task.name, task.pid, task.code))
            self.task_autoretry(task)
            task.set_stop_flag()
            return
        if task.stage == 1:
            log.info("Task: %s scheduled to second stage", task.name)
            self.run_task(TaskInfo(task.name, REGULAR_TASK, 0, task.priority), stage = 2)
        else:
            event_manager.emit(mirror.event.TaskStopEvent(task.name, task.pid, task.code))
            self.task_autoretry(task)
            task.set_stop_flag()
            task.stage = 1

    def task_autoretry(self, task):
        """
        If a task has a valid `autoretry`, and its interval is before next normal schedule,
        it will be used.

        """
        if task.autoretry <= 0:
            return
        if task.code == 0:
            return
        curtime   = int(time.time())
        next_time = task.get_schedule_time(since = curtime)
        if curtime + task.autoretry < next_time:
            taskinfo = self.queue[task.name]
            taskinfo.time = next_time
            self.reappend_task(task, taskinfo)

    def stop_all_tasks(self, signo = signal.SIGTERM):
        """
        This method can only be called when mirrord is shut down by SIGTERM or SIGINT.

        NOTE:
        Currently when mirrord is shut down, all running tasks will also be killed.

        """
        event_manager = component.get("EventManager")
        for taskname, task in self.tasks.items():
            if task.isinternal:
                continue
            if not task.running:
                continue
            pid = task.pid
            task.stop(signo)
            # Not sure it is ok...
            pid, status  = os.waitpid(pid, 0)

            endstr, code = self.parse_return_status(status)
            task.code    = code
            event_manager.emit(mirror.event.TaskStopEvent(task.name, task.pid, task.code))
            log.info("Killed task: %s with pid %d", taskname, pid)

    @classmethod
    def get_runnable_priority(cls, current, limit):
        """
        If limit is zero, all priority tasks can be run.
        Else if current value is lower than limit, all priority tasks can be run.
        Else it is a function between target priority and (current / limit).

        """
        if limit <= 0:

            return PRIORITY_MAX