Andrew/event loop fix (#72)

* create a new event loop for async data load

* create a new event loop for async data load if there is no active one

Author: AndrewNepogoda

workflow/__init__.py

@@ -12,5 +12,5 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-__version_info__ = (0, 0, 41)
+__version_info__ = (0, 0, 42)
 __version__ = ".".join(map(str, __version_info__))

workflow/executor.py

@@ -19,7 +19,7 @@
     FileMetadata
 from workflow.manager import ResourceManager
 from workflow.utils import save_csv_output, format_duration, build_models, extract_date_range, build_relation_path, \
-    get_common_model_relative_path
+    get_common_model_relative_path, get_or_create_eventloop
 
 
 class WorkflowStepState(str, Enum):
@@ -358,7 +358,7 @@ def _execute(self, logger: logging.Logger, env_config: EnvConfig, rai_config: Ra
             self.await_pending(env_config, logger, missed_resources)
 
     def await_pending(self, env_config, logger, missed_resources):
-        loop = asyncio.get_event_loop()
+        loop = get_or_create_eventloop()
         if loop.is_running():
             raise Exception('Waiting for resource would interrupt unexpected event loop - aborting to avoid confusion')
         pending = [src for src in missed_resources if self._resource_is_async(src)]

workflow/utils.py

@@ -194,3 +194,15 @@ def call_with_overhead(
     loop = asyncio.get_event_loop()
     loop.run_until_complete(
         call_with_overhead_async(f, logger, overhead_rate, start_time, timeout, max_tries, first_delay, max_delay))
+
+
+def get_or_create_eventloop():
+    """Get the current event loop or create a new one."""
+    try:
+        return asyncio.get_event_loop()
+    except RuntimeError as ex:
+        if "There is no current event loop in thread" in str(ex):
+            loop = asyncio.new_event_loop()
+            asyncio.set_event_loop(loop)
+            return asyncio.get_event_loop()
+        raise ex