WebHost: job pool based world generation

WebHostLib/autolauncher.py

@@ -4,8 +4,9 @@ import multiprocessing
 from datetime import timedelta, datetime
 import sys
 import typing
+import time
 
-from pony.orm import db_session, select
+from pony.orm import db_session, select, commit
 
 
 class CommonLocker():
@@ -23,7 +24,6 @@ class AlreadyRunningException(Exception):
 if sys.platform == 'win32':
     import os
 
-
     class Locker(CommonLocker):
         def __enter__(self):
             try:
@@ -42,7 +42,6 @@ if sys.platform == 'win32':
 else:  # unix
     import fcntl
 
-
     class Locker(CommonLocker):
         def __enter__(self):
             try:
@@ -66,24 +65,70 @@ def launch_room(room: Room, config: dict):
         multiworld.start()
 
 
-def autohost(config: dict):
-    import time
+def handle_generation_success(seed_id):
+    logging.info(f"Generation finished for seed {seed_id}")
 
+
+def handle_generation_failure(result: BaseException):
+    try:  # hacky way to get the full RemoteTraceback
+        raise result
+    except Exception as e:
+        logging.exception(e)
+
+
+def launch_generator(pool: multiprocessing.pool.Pool, generation: Generation):
+    @db_session
+    def handle_fail(result: BaseException):
+        generation.state = STATE_ERROR
+        handle_generation_failure(result)
+
+    logging.info(f"Generating {generation.id} for {len(generation.options)} players")
+
+    pool.apply_async(gen_game, (generation.options,),
+                     {"race": generation.meta["race"], "sid": generation.id, "owner": generation.owner},
+                     handle_generation_success, handle_generation_failure)
+    generation.state = STATE_STARTED
+
+
+def init_db(pony_config: dict):
+    db.bind(**pony_config)
+    db.generate_mapping()
+
+
+def autohost(config: dict):
     def keep_running():
         try:
             with Locker("autohost"):
-                logging.info("Starting autohost service")
-                # db.bind(**config["PONY"])
-                # db.generate_mapping(check_tables=False)
-                while 1:
-                    time.sleep(3)
-                    with db_session:
-                        rooms = select(
-                            room for room in Room if
-                            room.last_activity >= datetime.utcnow() - timedelta(days=3))
-                        for room in rooms:
-                            launch_room(room, config)
 
+                with multiprocessing.Pool(config["GENERATORS"], initializer=init_db,
+                                          initargs=(config["PONY"],)) as generator_pool:
+                    with db_session:
+                        to_start = select(generation for generation in Generation if generation.state == STATE_STARTED)
+
+                        if to_start:
+                            logging.info("Resuming generation")
+                            for generation in to_start:
+                                sid = Seed.get(id=generation.id)
+                                if sid:
+                                    generation.delete()
+                                else:
+                                    launch_generator(generator_pool, generation)
+
+                            commit()
+                        select(generation for generation in Generation if generation.state == STATE_ERROR).delete()
+
+                    while 1:
+                        time.sleep(0.50)
+                        with db_session:
+                            rooms = select(
+                                room for room in Room if
+                                room.last_activity >= datetime.utcnow() - timedelta(days=3))
+                            for room in rooms:
+                                launch_room(room, config)
+                            to_start = select(
+                                generation for generation in Generation if generation.state == STATE_QUEUED)
+                            for generation in to_start:
+                                launch_generator(generator_pool, generation)
         except AlreadyRunningException:
             pass
 
@@ -117,5 +162,6 @@ class MultiworldInstance():
             self.process = None
 
 
-from .models import Room
+from .models import Room, Generation, STATE_QUEUED, STATE_STARTED, STATE_ERROR, db, Seed
 from .customserver import run_server_process
+from .generate import gen_game