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
        if task.running and task.timeout > 0:
            return
        if not task.enabled:
            return
        taskinfo = TaskInfo(taskname, (SYSTEM_TASK if task.isinternal else REGULAR_TASK),
                            task.get_schedule_time(since), task.priority)

        # for system tasks and tasks-without-timeout-set, task is appended again
        # in run_*_task() method, for tasks-with-timeout-set, a timeout task
        # with same name is appended in run_task()
        if taskinfo in self.queue:
            return
        self.queue.put(taskinfo)
        event_manager = component.get("EventManager")
        event_manager.emit(mirror.event.TaskEnqueueEvent(taskname))

    def reappend_task(self, task, taskinfo):
        """
        Remove a taskinfo from queue and put it in again,
        to keep the queue in order.

        """
        if taskinfo not in self.queue:
            return
        self.queue.remove(taskinfo)
        self.queue.put(taskinfo)
        event_manager = component.get("EventManager")
        event_manager.emit(mirror.event.TaskEnqueueEvent(taskinfo.name))

    DEFAULT_BUFFER_SIZE  = 10240

    def write_mmap(self):
        data = pickle.dumps(self.queue)
        size = len(data) + 2 + 4
        if not hasattr(self, "buffer") or self.buffersz < size:
            self.buffersz = max(self.DEFAULT_BUFFER_SIZE, size)
            if hasattr(self, "buffer"):
                self.buffer.close()
            self.bufferfd = os.open("/tmp/mirrord",
                                    os.O_CREAT | os.O_TRUNC | os.O_RDWR,
                                    0o644)
            flag = fcntl.fcntl(self.bufferfd, fcntl.F_GETFD)
            fcntl.fcntl(self.bufferfd, fcntl.F_SETFD, flag | fcntl.FD_CLOEXEC)
            os.write(self.bufferfd, b'\x00' * self.buffersz)
            self.buffer   = mmap.mmap(self.bufferfd, self.buffersz, mmap.MAP_SHARED, mmap.PROT_WRITE)
            # close bufferfd
            os.close(self.bufferfd)
            self.buffer.write(b"\x79\x71")
        self.buffer.seek(2)
        self.buffer.write(struct.pack("I", size))
        self.buffer.write(data)

    def stop(self):
        log.info("Stopping mirror scheduler")
        if not hasattr(self, "buffer"):
            return
        self.buffer.close()
        os.unlink("/tmp/mirrord")

    def append_timeout_task(self, taskname, task, time):
        """
        A timeout checking task is added after a task begins to run.

        """
        if not task.running:
            return
        if not task.enabled:
            return
        taskinfo = TaskInfo(taskname, TIMEOUT_TASK,
                            time, task.priority)
        if taskinfo in self.queue:
            return
        self.queue.put(taskinfo)
        event_manager = component.get("EventManager")
        event_manager.emit(mirror.event.TaskEnqueueEvent(taskname))

    def remove_timeout_task(self, taskname):
        """
        It's slow...

        """
        taskqueue  = [ taskinfo for taskinfo in self.queue ]
        for taskinfo in taskqueue:
            if taskinfo.name == taskname and taskinfo.tasktype == TIMEOUT_TASK:
                self.queue.remove(taskinfo)
                return

    def init_general(self, config):
        self.emails    = []
        self.loadlimit = 4.0
        self.httpconn  = 1200
        self.logdir    = mirror.common.DEFAULT_TASK_LOG_DIR
        self.maxtasks  = 10

        if "general" not in config:
            log.error("Error in config file, no `general` section, will use default setting.")
            return
        emails = re.compile(r"([^@\s]+@[^@\s,]+)")
        emails = emails.findall(config['general']['emails'])
        for email in emails:
            self.emails.append(email)

        self.loadlimit = float(config['general']['loadlimit'])
        self.httpconn  = int  (config['general']['httpconn'] )
        self.maxtasks  = int  (config['general']['maxtasks'] )
        self.logdir    = config['general']['logdir']
        if self.logdir[-1] != os.path.sep:
            self.logdir += os.path.sep

    def reload_config(self):
        log.info("Stopping running tasks...")
        signal.signal(signal.SIGCHLD, signal.SIG_DFL)
        self.stop_all_tasks()
        signal.signal(signal.SIGCHLD, mirror.handler.sigchld_handler)

        log.info("Clearing old data...")
        self.tasks = odict(filter(lambda x:x[1].isinternal,self.tasks.items()))
        self.queue = Queue()

        log.info("Reloading new configs...")

        self.config = ConfigManager("mirror.ini", need_reload = True)