[Python知识库]python手写守护进程（daemon）

试图对抗服务器的排队机制：让一个 daemon 去排，排上了让它动态监测、调另外的 shell，并且就算目前所有 shell 都运行完， daemon 也保持一段等待时间才退出。在 shell 里面写要跑的命令，且可以多次动态加入，那只要 daemon 已开始、未退出，就可以随时加跑新任务，而不用重新排队。

要点：

• 固定输入 shell 的名字，如 zombie.sh；固定输入 shell 的摆放路径，即探测路径，如 ~/pool/。
• 输入 shell 要指定一个 flag，表示已经写完命令，可以执行，防止未敲完命令就被拿去执行。如约定 shell 最后一定以 echo DONE 命令结束，且只出现一次，则当且仅当检测到存在 zombie.sh，且里面有句 echo DONE，daemon 才拿它去执行。
• 为固定 shell 名字且避免命名冲突，执行 shell 之前改掉文件名先，如加一个序号；运行结束之后删掉相应 shell 文件，保持简洁。
• deamon 死循环监测（中间小 sleep 一阵，如 30s），且在所有 shell 都运行完之后，还保持一段等待时间（如半个钟），不然又要重新排；但又不好等太久，别被人投诉或者被管理员发现了。
• 每个 shell 文件都开一个线程执行，全部执行完之前 daemon 都不能停（无视等待时间超时），且每个 shell 执行完之后，都更新一下等待时间的开始计时时间（即 tic）。

Code

import os
import os.path as osp
import threading
import time
import re


DETECT_PATH = "./pool"
INPUT_SHELL = osp.join(DETECT_PATH, "zombie.sh")
RENAME_TEMPLATE = osp.join(DETECT_PATH, "zombie.{}.sh")
TIMEOUT = 20 * 60  # seconds
SLEEP = 30  # seconds


# 检测 `echo DONE`
fin_flag = re.compile(r"\necho\s+\"*DONE\"*")
# 等待时间的开始时间，要动态更新
tic, mutex_tic = time.time(), threading.Lock()
# 正在运行的 shell
n_running, mutex_run = 0, threading.Lock()
logger = open("log.txt", "w")
thread_list = []


def run():
    global tic, n_running, logger

    t = threading.currentThread()
    # tid = t.ident
    _n = int(t.name.replace("iTom-", ""))

    # starting log
    _timestamp = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
    _info_str = "START #{} at {}".format(_n, _timestamp)
    print(_info_str)
    logger.write(_info_str + "\n")
    logger.flush()

    # running counting +1
    mutex_run.acquire()
    n_running += 1
    mutex_run.release()

    new_f = RENAME_TEMPLATE.format(_n)
    os.system("bash {}".format(new_f))
    os.remove(new_f)

    # update global tic
    mutex_tic.acquire()
    tic = max(tic, time.time())
    mutex_tic.release()

    # running counting -1
    mutex_run.acquire()
    n_running -= 1
    mutex_run.release()

    # ending log
    _timestamp = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
    _info_str = "END #{} at {}".format(_n, _timestamp)
    print(_info_str)
    logger.write(_info_str + "\n")
    logger.flush()


while True:
    if osp.exists(INPUT_SHELL):
        try:
            with open(INPUT_SHELL, "r") as _f:
                _txt = _f.read()
            if len(fin_flag.findall(_txt)) > 0:
                # rename shell -> avoid name conflict
                _n = len(thread_list)
                new_f = RENAME_TEMPLATE.format(_n)
                os.system("mv {} {}".format(INPUT_SHELL, new_f))

                # run
                t = threading.Thread(target=run, name="iTom-{}".format(_n))
                thread_list.append(t)
                t.start()
                # t.join()  # NOT now, do it later

                # update tic
                mutex_tic.acquire()
                tic = max(tic, time.time())
                mutex_tic.release()
        except:
            pass

    toc = time.time()
    if (n_running < 1) and (toc - tic > TIMEOUT):
        logger.write("* TIME OUT\n")
        break

    time.sleep(SLEEP)

# 这句不知道还有没有必要
# 因为循环里面好像已经保证要等所有 shell 都运行完才结束
for t in thread_list:
    t.join()

logger.flush()
logger.close()

References

1. 多线程
2. Python 多线程
3. Python 日期和时间
4. 由bibtex生成引用文献字符串