from flask import Flask
'''
중간 생략
'''

# define the app
app = Flask(__name__)

'''
중간 생략
'''

# load module class
target_class = get_module_class(module_name, class_name, python_file_path)

logger.info("============== Start Serving Process ==============")

# create analyzer instance
instance = target_class(param, logger)

# load model
instance.init_serve()

# API route
@app.route("/serve", methods=["POST"])
def serve():
    try:
        start_time = time.time()
        _header, _body, errno, errmsg = instance.serving(header, body) # 서빙 함수 실행
        elapsed_time = time.time() - start_time
        logger.info(f"[End Serving] elapsed time : {str(elapsed_time)}")
    except Exception as exception:
        logger.exception(f"[Error] Unexpected exception during serving : {exception}")

'''
이하생략
'''

if __name__ == "__main__":
    logger.info("============== Run Flask app ==============")
    # This is used when started by server.
    app.run(host=host, port=int(port), debug=True, use_reloader=False)

import time
import ray

ray.init(num_cpus = 4)

# -----------------------------------------Bad Usage----------------------------------------------
@ray.remote
def tiny_work(x):
    time.sleep(0.0001) # Replace this with work you need to do.
    return x

start = time.time()
result_ids = [tiny_work.remote(x) for x in range(100000)]
results = ray.get(result_ids)
print("duration =", time.time() - start)

# ----------------------------------------Good Usage----------------------------------------------
def tiny_work(x):
    time.sleep(0.0001) # replace this is with work you need to do
    return x

@ray.remote
def mega_work(start, end):
    return [tiny_work(x) for x in range(start, end)]

start = time.time()
result_ids = []
[result_ids.append(mega_work.remote(x*1000, (x+1)*1000)) for x in range(100)]
results = ray.get(result_ids)
print("duration =", time.time() - start)

import time
import ray

ray.init(num_cpus = 4) # Specify this system has 4 CPUs.

@ray.remote
def do_some_work(x):
    time.sleep(1) # Replace this with work you need to do.
    return x

# -----------------------------------------Bad Usage----------------------------------------------
start = time.time()
bad_results = [ray.get(do_some_work.remote(x)) for x in range(4)]
print("duration =", time.time() - start)
print("results = ", bad_results )

# ----------------------------------------Good Usage----------------------------------------------
start = time.time()
good_results = ray.get([do_some_work.remote(x) for x in range(4)])
print("duration =", time.time() - start)
print("results = ", good_results)

import ray

# Start Ray. If you're connecting to an existing cluster, you would use
# ray.init(address=<cluster-address>) instead.
ray.init()

@ray.remote
class Counter(object):
    def __init__(self):
        self.value = 0

    def increment(self):
        self.value += 1
        return self.value

# Create ten Counter actors.
counters = [Counter.remote() for _ in range(10)]

# Increment each Counter once and get the results. These tasks all happen in
# parallel.
results = ray.get([c.increment.remote() for c in counters])
print(results)  # prints [1, 1, 1, 1, 1, 1, 1, 1, 1, 1]

# Increment the first Counter five times. These tasks are executed serially
# and share state.
results = ray.get([counters[0].increment.remote() for _ in range(5)])
print(results)  # prints [2, 3, 4, 5, 6]

import ray

@ray.remote
class Algorithm(aimodel.AIModel):
    def __init__(self, name, config, logger):
        self.config = config
        self.logger = logger
        '''
        이하 생략
        '''

    def predict_by_ray(self, input_param):
        return [self.predict(*x) for x in input_param]


    def predict(self, **parameter,  fail_type=None):
        self.logger.debug(f"input data: {target}:\n{serving_time}"
        '''
        이하 생략
        '''

'''
생략
'''
import ray

ray.init()

class Analyzer(aimodule.AIModule):
    def __init__(self, config, logger):
        self.config = config
        self.logger = logger

        # Ray Actor 개수 설정
        self.number_of_ray_actor = int(psutil.cpu_count(logical=False) * 0.3)

        # Ray Actor Pool 생성
        self.serving_algorithm_actor_pool = [Algorithm.remote(self.service_id, self.config, self.logger) 
for i in range(self.number_of_ray_actor)]

        '''
        이하 생략
        '''
    def serve(self, header, data_dict):
        # API에서 호출되는 서빙 함수
        self.logger.info("=========== Start Serving ===========")
        '''
        중간 생략
        '''
        # 기존 순차 처리 방식에 필요한 정보를 모은다.
        test_target_id = list(input_df.index.values)[0]
        test_count = 500
        input_param = [(test_target_id, serving_time, df, 'top') for i in range(test_count)]

        # Actor 개수에 맞게 작업들을 나눈다.
        chunk_size, remainder = divmod(len(input_param), self.number_of_ray_actor)
        splitted_input_param = [input_param[i : (i + chunk_size)] for i in range(0, len(input_param), chunk_size)]
        task_per_actor = zip(splitted_input_param, self.serving_algorithm_actor_pool)

        # Task 파라미터에 담아 remote task를 호출한다.
        with TimeLogger("ray serving takes : ", self.logger):
            ray_serving_result = ray.get([actor.predict_by_ray.remote(task) for task, actor in task_per_actor])
            self.logger.info(f"ray serving result : {ray_serving_result}")