py_loader: implement metacall_await in the python port

source/loaders/py_loader/source/py_loader_port.c

@@ -673,8 +673,63 @@ static PyObject *py_loader_port_invoke(PyObject *self, PyObject *var_args)
 	return result;
 }
 
-// TODO
-#if 0
+static void *py_loader_port_await_settle(void *result, void *data, int rejected)
+{
+	loader_impl impl = loader_get_impl(py_loader_tag);
+	PyObject *future = (PyObject *)data;
+	PyObject *py_result, *ret;
+
+	py_loader_thread_acquire();
+
+	py_result = py_loader_impl_value_to_capi(impl, value_type_id(result), result);
+
+	if (py_result == NULL)
+	{
+		PyErr_Clear();
+		py_result = Py_ReturnNone();
+	}
+
+	if (rejected == 0)
+	{
+		ret = PyObject_CallMethod(future, "set_result", "O", py_result);
+	}
+	else
+	{
+		/* set_exception requires an exception instance, so wrap the rejected value */
+		PyObject *exception = PyExceptionInstance_Check(py_result) ? py_result : PyObject_CallFunctionObjArgs(PyExc_RuntimeErrorPtr(), py_result, NULL);
+
+		ret = PyObject_CallMethod(future, "set_exception", "O", exception);
+
+		if (exception != py_result)
+		{
+			Py_DecRef(exception);
+		}
+	}
+
+	if (ret == NULL)
+	{
+		PyErr_Print();
+	}
+
+	Py_DecRef(ret);
+	Py_DecRef(py_result);
+	Py_DecRef(future);
+
+	py_loader_thread_release();
+
+	return NULL;
+}
+
+static void *py_loader_port_await_resolve(void *result, void *data)
+{
+	return py_loader_port_await_settle(result, data, 0);
+}
+
+static void *py_loader_port_await_reject(void *result, void *data)
+{
+	return py_loader_port_await_settle(result, data, 1);
+}
+
 static PyObject *py_loader_port_await(PyObject *self, PyObject *var_args)
 {
 	PyObject *name, *result = NULL;
@@ -694,24 +749,24 @@ static PyObject *py_loader_port_await(PyObject *self, PyObject *var_args)
 
 	if (var_args_size == 0)
 	{
-		PyErr_SetString(PyExc_TypeErrorPtr(), "Invalid number of arguments, use it like: metacall('function_name', 'asd', 123, [7, 4]);");
+		PyErr_SetString(PyExc_TypeErrorPtr(), "Invalid number of arguments, use it like: metacall_await('function_name', 'asd', 123, [7, 4]);");
 		return Py_ReturnNone();
 	}
 
 	name = PyTuple_GetItem(var_args, 0);
 
-	#if PY_MAJOR_VERSION == 2
+#if PY_MAJOR_VERSION == 2
 	{
 		if (!(PyString_Check(name) && PyString_AsStringAndSize(name, &name_str, &name_length) != -1))
 		{
 			name_str = NULL;
 		}
 	}
-	#elif PY_MAJOR_VERSION == 3
+#elif PY_MAJOR_VERSION == 3
 	{
 		name_str = PyUnicode_Check(name) ? (char *)PyUnicode_AsUTF8AndSize(name, &name_length) : NULL;
 	}
-	#endif
+#endif
 
 	if (name_str == NULL)
 	{
@@ -744,55 +799,65 @@ static PyObject *py_loader_port_await(PyObject *self, PyObject *var_args)
 
 	/* Execute the await */
 	{
+		PyObject *futures_module = PyImport_ImportModule("concurrent.futures");
 		void *ret;
 
-		py_loader_thread_release();
-
-		/* TODO: */
-		/*
-		if (value_args != NULL)
+		if (futures_module == NULL)
 		{
-			ret = metacallv_s(name_str, value_args, args_size);
-		}
-		else
-		{
-			ret = metacallv_s(name_str, metacall_null_args, 0);
+			goto clear;
 		}
-		*/
 
-		py_loader_thread_acquire();
+		/* The caller blocks on future.result() while the loader settles it from another thread */
+		result = PyObject_CallMethod(futures_module, "Future", NULL);
+		Py_DecRef(futures_module);
 
-		if (ret == NULL)
+		if (result == NULL)
 		{
-			result = Py_ReturnNone();
 			goto clear;
 		}
 
-		result = py_loader_impl_value_to_capi(impl, value_type_id(ret), ret);
+		/* This reference is owned by the callbacks, they release it once the future is settled */
+		Py_IncRef(result);
 
-		value_type_destroy(ret);
+		py_loader_thread_release();
 
-		if (result == NULL)
+		ret = metacall_await_s(name_str, value_args != NULL ? value_args : metacall_null_args, args_size, &py_loader_port_await_resolve, &py_loader_port_await_reject, result);
+
+		py_loader_thread_acquire();
+
+		if (ret == NULL)
 		{
-			result = Py_ReturnNone();
+			/* The callbacks will not be called, release both references */
+			Py_DecRef(result);
+			Py_DecRef(result);
+			result = NULL;
+			PyErr_Format(PyExc_RuntimeErrorPtr(), "Failed to await the function '%s'", name_str);
 			goto clear;
 		}
+
+		/* The future is settled through the callbacks, so the returned value is not needed */
+		py_loader_thread_release();
+		value_type_destroy(ret);
+		py_loader_thread_acquire();
 	}
 
 clear:
 	if (value_args != NULL)
 	{
+		py_loader_thread_release();
+
 		for (args_count = 0; args_count < args_size; ++args_count)
 		{
 			value_type_destroy(value_args[args_count]);
 		}
 
+		py_loader_thread_acquire();
+
 		free(value_args);
 	}
 
 	return result;
 }
-#endif
 
 static PyObject *py_loader_port_inspect(PyObject *self, PyObject *args)
 {
@@ -1036,6 +1101,8 @@ static PyMethodDef metacall_methods[] = {
 		"Get information about all loaded objects." },
 	{ "metacall", py_loader_port_invoke, METH_VARARGS,
 		"Call a function anonymously." },
+	{ "metacall_await", py_loader_port_await, METH_VARARGS,
+		"Call an async function anonymously, returns a concurrent.futures.Future." },
 	{ "metacall_value_create_ptr", py_loader_port_value_create_ptr, METH_VARARGS,
 		"Create a new value of type Pointer." },
 	{ "metacall_value_reference", py_loader_port_value_reference, METH_VARARGS,

source/ports/py_port/metacall/__init__.py

@@ -17,4 +17,4 @@
 #	See the License for the specific language governing permissions and
 #	limitations under the License.
 
-from metacall.api import metacall, metacall_load_from_file, metacall_load_from_file_export, metacall_load_from_memory, metacall_load_from_package, metacall_load_from_package_ex, metacall_inspect, metacall_value_create_ptr, metacall_value_reference, metacall_value_dereference
+from metacall.api import metacall, metacall_await, metacall_load_from_file, metacall_load_from_file_export, metacall_load_from_memory, metacall_load_from_package, metacall_load_from_package_ex, metacall_inspect, metacall_value_create_ptr, metacall_value_reference, metacall_value_dereference

source/ports/py_port/metacall/api.py

@@ -135,6 +135,10 @@ def metacall_load_from_memory(tag, buffer):
 def metacall(function_name, *args):
 	return module.metacall(function_name, *args)
 
+# Async invocation, returns a concurrent.futures.Future
+def metacall_await(function_name, *args):
+	return module.metacall_await(function_name, *args)
+
 # Wrap metacall inspect and transform the json string into a dict
 def metacall_inspect():
 	data = module.metacall_inspect()

source/tests/CMakeLists.txt

@@ -184,7 +184,7 @@ add_subdirectory(metacall_python_relative_path_test)
 add_subdirectory(metacall_python_without_functions_test)
 add_subdirectory(metacall_python_builtins_test)
 add_subdirectory(metacall_python_async_test)
-# TODO: add_subdirectory(metacall_python_await_test) # TODO: Implement metacall_await in Python Port
+add_subdirectory(metacall_python_await_test)
 add_subdirectory(metacall_python_exception_test)
 # TODO: add_subdirectory(metacall_python_node_await_test) # TODO: Implement metacall_await in Python Port
 add_subdirectory(metacall_python_without_env_vars_test)

source/tests/metacall_python_await_test/source/metacall_python_await_test.cpp

@@ -39,30 +39,17 @@ TEST_F(metacall_python_await_test, DefaultConstructor)
 #if defined(OPTION_BUILD_LOADERS_PY)
 	{
 		static const char buffer[] =
-			"import asyncio\n"
 			"import sys\n"
 			"sys.path.insert(0, '" METACALL_PYTHON_PORT_PATH "')\n"
 			"from metacall import metacall_load_from_memory, metacall_await\n"
 			"script = \"\"\"\n"
 			"async def yeet(n):\n"
 			"	return n\n"
 			"\"\"\"\n"
-			"result = 0\n"
-			"lock = threading.Lock()\n"
 			"metacall_load_from_memory('py', script)\n"
-			"async def test():\n"
-			"	result = await metacall_await('yeet', 1234)\n"
-			"	if result != 1234:\n"
-			"		sys.exit(1)\n"
-			"	else:\n"
-			"		lock.acquire()\n"
-			"		result = 1\n"
-			"		lock.release()\n"
-			"asyncio.run(test())\n"
-			"lock.acquire()\n"
-			"if result != 1:\n"
-			"	sys.exit(2)\n"
-			"lock.release()\n";
+			"f = metacall_await('yeet', 1234)\n"
+			"if f.result(timeout=10) != 1234:\n"
+			"	sys.exit(1)\n";
 
 		ASSERT_EQ((int)0, (int)metacall_load_from_memory("py", buffer, sizeof(buffer), NULL));
 	}