from multiprocess import Process

process = Process()

from multiprocess import Process
import os
import math

def work_fun(work_list):
    pass

def distrib_works(work_list, process_num):
    group_length = math.ceil(len(filename_list) / process_num)
    return [work_list[(i*group_length):((i+1)*group_length)] for i in range(process_num)]


work_list = os.listdir("../data")
process_num = 4
group = distrib_works(work_list, process_num)

process_list = [Process(target=work_fun, args=(g,)) for g in group_list]
for p in process_list:
    p.daemon = True
    p.start()
for p in process_list:
    p.join()

from multiprocess import Process

class MyProcess(Process):
    position = 0
    works = None

    def __init__(self, position, works)
        Process.__init__(self)
        self.position = poisition
        self.works = works

    def run():
        pass

from multiprocess import Process, Lock

def distrib_works(work_list: List[str], process_num) -> List[List[str]]:
    group_length = math.ceil(len(work_list) / process_num)
    return [g for g in [work_list[(i*group_length):((i+1)*group_length)] for i in range(process_num)] if len(g) > 0]

class FindTargetTaxiProcess(multiprocessing.Process):

    def __init__(self, input_files, index, lock, log_file):
        multiprocessing.Process.__init__(self, target=find_target, args=(lock, log_file))
        self.input_files = input_files
        self.index = index

    def find_target(self, lock, log_file):
        for filename in self.input_files:
            with open(filename) as in_file:
                for row in tqdm(in_file, ncols=80, position=self.index):
                    cells = row.split(",")
                    if int(cells[0]) == 11865:
                        print(row)

if __name__ == '__main__':
    LOG_FILE = r"E:\出租车点\上下车点\scripts\data\find_error.log"
    lock = multiprocessing.Lock()
    # ROOT_DIR = "../data/201502/temp"
    ROOT_DIR = r"E:\出租车点\201502\RawCSV"
    INPUT_FILES = [os.path.join(ROOT_DIR, f) for f in os.listdir(ROOT_DIR)]
    GROUP_LIST = distrib_works(INPUT_FILES, 4)
    PROCESS_LIST = [FineTargetTaxiProcess(element, i, lock) for i, element in enumerate(GROUP_LIST)]
    for process in PROCESS_LIST:
        process.start()

from multiprocess import Pool

p = Pool(process=6)

from multiprocess import Process, Pool, Lock

def find_target(in_file, lock, log_file):
    with open(filename) as in_file:
        for row in tqdm(in_file, ncols=80, position=self.index):
            cells = row.split(",")
            if int(cells[0]) == 11865:
                with lock:
                    with open(log_file, mode="a") as log:
                        print(row, file=log)

if __name__ == '__main__':
    LOCK = multiprocessing.Lock()
    LOG_FILE = r"E:\出租车点\上下车点\scripts\data\find_error.log"
    ROOT_DIR = r"E:\出租车点\201502\RawCSV"
    INPUT_FILES = [os.path.join(ROOT_DIR, f) for f in os.listdir(ROOT_DIR)]
    POOL = Pool(process=6)
    for f in INPUT_FILES:
        POOL.apply_async(find_target, (f, LOCK, LOG_FILE))
    POOL.close()
    POOL.join()

def run(self):
    for filename in self.input_files:
        with open(filename, encoding="GB2312") as in_file:
            for row in tqdm(in_file):
                cells = row.split(",")
                if int(cells[0]) == 11865:
                    with self.lock:
                        with open(TARGET_TAXI_FILE, mode="a") as log:
                            print(row, file=log)

import multiprocess


class Producer(multiprocessing.Process):
    def __init__(self, productList, condition):
        multiprocessing.Process.__init__(self)
        self.productList = productList  # type: List
        self.condition = condition  # type: multiprocess.Condition

    def run(self):
        while True:
            product = random.randint(0, 100)
            with self.condition:
                print("条件锁：被 生产者 获取")
                self.productList.append(product)
                print(f"生产者：产生了 {product}。")
                print("生产者：唤醒消费者线程")
                self.condition.notify()
                print("条件锁：被 生产者 释放")
            time.sleep(1)


class Customer(multiprocessing.Process):

    def __init__(self, productList, condition):
        multiprocessing.Process.__init__(self)
        self.productList = productList  # type: List
        self.condition = condition  # type: multiprocess.Condition

    def run(self):
        while True:
            with self.condition:
                print("条件锁：被 消费者 获取")
                while True:
                    if self.productList:
                        product = self.productList.pop()
                        print(f"消费者：消费了 {product}")
                        break
                    print("消费者：等待生产者")
                    self.condition.wait()
                print("条件锁：被 消费者 释放")


def main():
    manager = multiprocessing.Manager()
    productList = manager.list()
    condition = multiprocessing.Condition()
    process_producer = Producer(productList, condition)
    process_customer = Customer(productList, condition)
    process_producer.start()
    process_customer.start()
    process_producer.join()
    process_customer.join()

if __name__ == '__main__':
    main()

from multiprocessing import Process, current_process, Semaphore
import time

def worker(s, i):
    s.acquire()
    print(current_process().name + "acquire");
    time.sleep(i)
    print(current_process().name + "release\n");
    s.release()

if __name__ == "__main__":
    s = Semaphore(2)
    for i in range(5):
        p = Process(target = worker, args=(s, i*2))
        p.start()

from multiprocessing import Process, Lock, Manager
import time

def work(lock, var, index):
    with lock:
        var.append(index)
        print(f"Process {index} apped {index}")

if __name__ == '__main__':
    var = Manager().list()
    lock = Lock()
    process_list = [Process(target=work, args=(lock, var, i)) for i in range(8)]
    for p in process_list:
        p.start()
    for p in process_list:
        p.join()
    print(var)

import multiprocessing
import time
def wait_for_event(e):
    print("wait_for_event: starting")
    e.wait()
    print("wairt_for_event: e.is_set()->" + str(e.is_set()))

def wait_for_event_timeout(e, t):
    print("wait_for_event_timeout:starting")
    e.wait(t)
    print("wait_for_event_timeout:e.is_set->" + str(e.is_set()))

if __name__ == "__main__":
    e = multiprocessing.Event()
    w1 = multiprocessing.Process(target=wait_for_event, args=(e,))
    w2 = multiprocessing.Process(target=wait_for_event_timeout, args=(e, 6))
    w1.start()
    w2.start()
    time.sleep(10)
    print("main: event setting")
    e.set()
    print("main: event is set")

class FineTargetTaxiProcess(mp.Process):
    '''
    处理进程：多进程方式处理文件，结果全部传递给打印进程。
    '''
    def __init__(self, input_files, index, queue):
        mp.Process.__init__(self, target=self.pick, args=(queue,))
        self.input_files = input_files
        self.index = index
        self.lock = lock

    def pick(self, queue):
        for filename in tqdm(self.input_files, ncols=80, position=self.index, desc=f"Process {self.index}"):
            with open(filename, encoding="GB2312") as in_file:
                for row in in_file:
                    cells = row.split(",")
                    if int(cells[0]) == 11865:
                        try:
                            queue.put(",".join(cells), block=False)
                        except:
                            print("Queue full")

class PrinterProcess(mp.Process):
    '''
    打印进程：维持对输出文件的打开状态，打印数据。
    可避免频繁打开、关闭结果文件造成的系统开销，
    但是引入了消息传递的开销。
    '''
    def __init__(self, output_file, log, queue):
        mp.Process.__init__(self, target=self.write, args=(queue,))
        self.output_file = output_file
        self.log_file = log

    def write(self, queue):
        with open(self.output_file, mode="w", newline="\n") as printer, open(self.log_file, mode="w") as log:
            while True:
                try:
                    row = queue.get(block=True, timeout=1)
                    print(row, file=printer)
                except:
                    print("Queue empty", file=log)

if __name__ == '__main__':
    lock = mp.Lock()
    ROOT_DIR = r"/mnt/e/出租车点/201502/RawCSV"
    INPUT_FILES = [os.path.join(ROOT_DIR, f) for f in os.listdir(ROOT_DIR)]
    GROUP_LIST = distrib_works(INPUT_FILES, 6)
    QUEUE = mp.Queue()
    PROCESS_LIST = [FineTargetTaxiProcess(element, i, QUEUE) for i, element in enumerate(GROUP_LIST)]
    PRINTER_PROCESS = PrinterProcess("./data/usequeue.txt", "./data/usequeue.log", QUEUE)
    for process in PROCESS_LIST:
        process.daemon = True
        process.start()
    PRINTER_PROCESS.daemon = True
    PRINTER_PROCESS.start()
    for p in PROCESS_LIST:
        p.join()

class FineTargetTaxiProcess(mp.Process):
    '''
    处理进程：多进程方式处理文件，结果全部传递给打印进程。
    '''
    def __init__(self, input_files, index, pipe):
        mp.Process.__init__(self, target=self.pick, args=(pipe,))
        self.input_files = input_files
        self.index = index
        self.lock = lock

    def pick(self, pipe):
        for filename in tqdm(self.input_files, ncols=80, position=self.index, desc=f"Process {self.index}"):
            with open(filename, encoding="GB2312") as in_file:
                for row in in_file:
                    cells = row.split(",")
                    if int(cells[0]) == 11865:
                        try:
                            pipe.send(",".join(cells))
                        except e as Exception:
                            print("Pipe send error")

class PrinterProcess(mp.Process):
    '''
    打印进程：维持对输出文件的打开状态，打印数据。
    可避免频繁打开、关闭结果文件造成的系统开销，
    但是引入了消息传递的开销。
    '''
    def __init__(self, output_file, log, pipe):
        mp.Process.__init__(self, target=self.write, args=(pipe,))
        self.output_file = output_file
        self.log_file = log

    def write(self, pipe):
        with open(self.output_file, mode="w", newline="\n") as printer, open(self.log_file, mode="w") as log:
            while True:
                try:
                    row = pipe.recv()
                    print(row, file=printer)
                except e as Exception:
                    print("Pipe read error", file=log)

if __name__ == '__main__':
    lock = mp.Lock()
    # ROOT_DIR = "../data/201502/temp"
    ROOT_DIR = r"/mnt/e/出租车点/201502/RawCSV"
    INPUT_FILES = [os.path.join(ROOT_DIR, f) for f in os.listdir(ROOT_DIR)]
    GROUP_LIST = distrib_works(INPUT_FILES, 6)
    (RECEIVER, SENDER) = mp.Pipe()
    PROCESS_LIST = [FineTargetTaxiProcess(element, i, SENDER) for i, element in enumerate(GROUP_LIST)]
    PRINTER_PROCESS = PrinterProcess("./data/usepipe.txt", "./data/usepipe.log", RECEIVER)
    for process in PROCESS_LIST:
        process.daemon = True
        process.start()
    PRINTER_PROCESS.daemon = True
    PRINTER_PROCESS.start()
    for p in PROCESS_LIST:
        p.join()