fix(rpc_loader): creating per-call curl handler for sync invocation for thread-safety (#765)

* test: add synchronous concurrent producers test for RPC calls

* refactor: improve Sync calls' thread safety and error handling in RPC calls

Author: Mag-D-Anas

source/loaders/rpc_loader/source/rpc_loader_impl.cpp

@@ -65,7 +65,6 @@ struct rpc_async_context;
 typedef struct loader_impl_rpc_type
 {
 	CURL *discover_curl;
-	CURL *invoke_curl;
 	CURLM *async_multi;
 	std::thread poll_thread;
 	std::atomic<bool> exit_flag;
@@ -195,22 +194,39 @@ function_return function_rpc_interface_invoke(function func, function_impl impl,
 		return NULL;
 	}
 
-	/* Execute a POST to the endpoint */
+	/* This creates a per-call CURL handle for thread safety */
+	CURL *easy = curl_easy_init();
+
+	if (easy == NULL)
+	{
+		log_write("metacall", LOG_LEVEL_ERROR, "Could not create CURL handle for sync call to %s", rpc_function->url.c_str());
+		metacall_allocator_free(rpc_impl->allocator, buffer);
+		return NULL;
+	}
+
 	loader_impl_rpc_write_data_type write_data;
 
-	curl_easy_setopt(rpc_impl->invoke_curl, CURLOPT_URL, rpc_function->url.c_str());
-	curl_easy_setopt(rpc_impl->invoke_curl, CURLOPT_POSTFIELDS, buffer);
-	curl_easy_setopt(rpc_impl->invoke_curl, CURLOPT_POSTFIELDSIZE, body_request_size - 1);
-	curl_easy_setopt(rpc_impl->invoke_curl, CURLOPT_WRITEDATA, static_cast<loader_impl_rpc_write_data>(&write_data));
+	curl_easy_setopt(easy, CURLOPT_VERBOSE, CURL_VERBOSE);
+	curl_easy_setopt(easy, CURLOPT_HEADER, 0L);
+	curl_easy_setopt(easy, CURLOPT_CUSTOMREQUEST, "POST");
+	curl_easy_setopt(easy, CURLOPT_HTTPHEADER, rpc_impl->headers);
+	curl_easy_setopt(easy, CURLOPT_USERAGENT, "librpc_loader/0.1");
+	curl_easy_setopt(easy, CURLOPT_WRITEFUNCTION, rpc_loader_impl_write_data);
+	curl_easy_setopt(easy, CURLOPT_URL, rpc_function->url.c_str());
+	curl_easy_setopt(easy, CURLOPT_POSTFIELDS, buffer);
+	curl_easy_setopt(easy, CURLOPT_POSTFIELDSIZE, body_request_size - 1);
+	curl_easy_setopt(easy, CURLOPT_WRITEDATA, static_cast<loader_impl_rpc_write_data>(&write_data));
+
+	CURLcode res = curl_easy_perform(easy);
 
-	CURLcode res = curl_easy_perform(rpc_impl->invoke_curl);
+	curl_easy_cleanup(easy);
 
 	/* Clear the request buffer */
 	metacall_allocator_free(rpc_function->rpc_impl->allocator, buffer);
 
 	if (res != CURLE_OK)
 	{
-		log_write("metacall", LOG_LEVEL_ERROR, "Could not call to the API endpoint %s [%]", rpc_function->url.c_str(), curl_easy_strerror(res));
+		log_write("metacall", LOG_LEVEL_ERROR, "Could not call to the API endpoint %s [%s]", rpc_function->url.c_str(), curl_easy_strerror(res));
 		return NULL;
 	}
 
@@ -541,34 +557,11 @@ loader_impl_data rpc_loader_impl_initialize(loader_impl impl, configuration conf
 	curl_easy_setopt(rpc_impl->discover_curl, CURLOPT_HEADER, 0L);
 	curl_easy_setopt(rpc_impl->discover_curl, CURLOPT_WRITEFUNCTION, rpc_loader_impl_write_data);
 
-	/* Initialize invoke CURL object */
-	rpc_impl->invoke_curl = curl_easy_init();
-
-	if (rpc_impl->invoke_curl == NULL)
-	{
-		log_write("metacall", LOG_LEVEL_ERROR, "Could not create CURL invoke object");
-
-		curl_easy_cleanup(rpc_impl->discover_curl);
-
-		metacall_allocator_destroy(rpc_impl->allocator);
-
-		delete rpc_impl;
-
-		return NULL;
-	}
-
 	rpc_impl->headers = NULL;
 	rpc_impl->headers = curl_slist_append(rpc_impl->headers, "Accept: application/json");
 	rpc_impl->headers = curl_slist_append(rpc_impl->headers, "Content-Type: application/json");
 	rpc_impl->headers = curl_slist_append(rpc_impl->headers, "charset: utf-8");
 
-	curl_easy_setopt(rpc_impl->invoke_curl, CURLOPT_VERBOSE, CURL_VERBOSE);
-	curl_easy_setopt(rpc_impl->invoke_curl, CURLOPT_HEADER, 0L);
-	curl_easy_setopt(rpc_impl->invoke_curl, CURLOPT_CUSTOMREQUEST, "POST");
-	curl_easy_setopt(rpc_impl->invoke_curl, CURLOPT_HTTPHEADER, rpc_impl->headers);
-	curl_easy_setopt(rpc_impl->invoke_curl, CURLOPT_USERAGENT, "librpc_loader/0.1");
-	curl_easy_setopt(rpc_impl->invoke_curl, CURLOPT_WRITEFUNCTION, rpc_loader_impl_write_data);
-
 	/* Initialize async multi handle */
 	rpc_impl->async_multi = curl_multi_init();
 
@@ -577,7 +570,6 @@ loader_impl_data rpc_loader_impl_initialize(loader_impl impl, configuration conf
 		log_write("metacall", LOG_LEVEL_ERROR, "Could not create CURL multi handle for async");
 
 		curl_easy_cleanup(rpc_impl->discover_curl);
-		curl_easy_cleanup(rpc_impl->invoke_curl);
 		metacall_allocator_destroy(rpc_impl->allocator);
 		delete rpc_impl;
 
@@ -598,7 +590,6 @@ loader_impl_data rpc_loader_impl_initialize(loader_impl impl, configuration conf
 
 		curl_multi_cleanup(rpc_impl->async_multi);
 		curl_easy_cleanup(rpc_impl->discover_curl);
-		curl_easy_cleanup(rpc_impl->invoke_curl);
 		metacall_allocator_destroy(rpc_impl->allocator);
 		delete rpc_impl;
 
@@ -937,11 +928,9 @@ int rpc_loader_impl_destroy(loader_impl impl)
 
 	curl_easy_cleanup(rpc_impl->discover_curl);
 
-	curl_easy_cleanup(rpc_impl->invoke_curl);
-
 	curl_global_cleanup();
 
 	delete rpc_impl;
 
 	return 0;
-}
+}

source/tests/metacall_rpc_test/source/metacall_rpc_test.cpp

@@ -286,6 +286,87 @@ static const int NUM_THREADS = 4;
 static const int CALLS_PER_THREAD = 10;
 static const int TOTAL_CONCURRENT = NUM_THREADS * CALLS_PER_THREAD;
 
+TEST_F(metacall_rpc_test, SyncConcurrentProducers)
+{
+	ASSERT_EQ((int)0, (int)metacall_initialize());
+
+#if defined(OPTION_BUILD_LOADERS_RPC)
+	{
+		const char *rpc_scripts[] = { "remote.url" };
+		void *handle = NULL;
+
+		ASSERT_EQ((int)0, (int)metacall_load_from_file("rpc", rpc_scripts, 1, &handle));
+
+		ASSERT_NE((void *)NULL, (void *)handle);
+
+		std::atomic<int> sync_failures(0);
+		std::atomic<int> sync_mismatches(0);
+		std::vector<std::thread> threads;
+
+		for (int t = 0; t < NUM_THREADS; t++)
+		{
+			threads.emplace_back([t, handle, &sync_failures, &sync_mismatches]() {
+				for (int i = 0; i < CALLS_PER_THREAD; i++)
+				{
+					float a = static_cast<float>(t * 100 + i + 1);
+					float b = 1.0f;
+					float expected = a / b;
+
+					const enum metacall_value_id divide_ids[] = {
+						METACALL_FLOAT, METACALL_FLOAT
+					};
+
+					void *ret = metacallht_s(handle, "divide", divide_ids, 2, a, b);
+
+					if (ret == NULL)
+					{
+						std::cout << "    [SYNC FAIL] thread=" << t
+								  << " call=" << i
+								  << " ret=NULL" << std::endl;
+						sync_failures.fetch_add(1);
+						continue;
+					}
+
+					double actual = extract_numeric(ret);
+
+					if (actual < expected - 0.01 || actual > expected + 0.01)
+					{
+						std::cout << "    [SYNC MISMATCH] thread=" << t
+								  << " call=" << i
+								  << " expected=" << expected
+								  << " got=" << actual << std::endl;
+						sync_mismatches.fetch_add(1);
+					}
+
+					metacall_value_destroy(ret);
+				}
+
+				std::cout << "Thread " << t << ": completed "
+						  << CALLS_PER_THREAD << " sync calls" << std::endl;
+			});
+		}
+
+		for (auto &th : threads)
+		{
+			th.join();
+		}
+
+		int total_calls = NUM_THREADS * CALLS_PER_THREAD;
+
+		std::cout << "SyncConcurrentProducers: total=" << total_calls
+				  << ", failures=" << sync_failures.load()
+				  << ", mismatches=" << sync_mismatches.load() << std::endl;
+
+		EXPECT_EQ(sync_failures.load(), 0) << "All sync calls should succeed";
+		EXPECT_EQ(sync_mismatches.load(), 0) << "All sync results should match expected values";
+
+		EXPECT_EQ((int)0, (int)metacall_clear(handle));
+	}
+#endif
+
+	metacall_destroy();
+}
+
 TEST_F(metacall_rpc_test, AsyncConcurrentProducers)
 {
 	ASSERT_EQ((int)0, (int)metacall_initialize());
@@ -470,86 +551,4 @@ static void *on_error_reject(void *error, void * /*data*/)
 
 	g_error_rejected.fetch_add(1);
 	return NULL;
-}
-
-static const int GOOD_CALLS = 5;
-
-// TODO: It fails under address sanitizer and thread sanitizer
-TEST_F(metacall_rpc_test, ErrorUnderConcurrency)
-{
-	ASSERT_EQ((int)0, (int)metacall_initialize());
-
-#if defined(OPTION_BUILD_LOADERS_RPC)
-	{
-		const char *rpc_scripts[] = { "remote.url" };
-		ASSERT_EQ((int)0, (int)metacall_load_from_file("rpc", rpc_scripts, 1, NULL));
-
-		g_error_resolved.store(0);
-		g_error_rejected.store(0);
-
-		call_context good_contexts[GOOD_CALLS];
-
-		for (int i = 0; i < GOOD_CALLS; i++)
-		{
-			float a = static_cast<float>((i + 1) * 10);
-			float b = static_cast<float>(i + 1);
-
-			good_contexts[i].call_id = i;
-			good_contexts[i].expected = static_cast<double>(a / b);
-
-			void *args[] = {
-				metacall_value_create_float(a),
-				metacall_value_create_float(b)
-			};
-
-			metacall_await_s("async_divide", args, 2,
-				on_error_resolve, on_error_reject, &good_contexts[i]);
-
-			metacall_value_destroy(args[0]);
-			metacall_value_destroy(args[1]);
-		}
-
-		{
-			void *bad_args[] = {
-				metacall_value_create_int(1),
-				metacall_value_create_int(2)
-			};
-
-			metacall_await_s("sum", bad_args, 2,
-				on_error_resolve, on_error_reject, NULL);
-
-			metacall_value_destroy(bad_args[0]);
-			metacall_value_destroy(bad_args[1]);
-		}
-
-		int total_expected = GOOD_CALLS + 1;
-		bool reached = false;
-		int waited = 0;
-
-		while (waited < 10000)
-		{
-			int total = g_error_resolved.load() + g_error_rejected.load();
-
-			if (total >= total_expected)
-			{
-				reached = true;
-				break;
-			}
-
-			std::this_thread::sleep_for(std::chrono::milliseconds(10));
-			waited += 10;
-		}
-
-		std::cout << "ErrorUnderConcurrency: Resolved=" << g_error_resolved.load()
-				  << ", Rejected=" << g_error_rejected.load() << std::endl;
-
-		EXPECT_TRUE(reached) << "All callbacks (good + bad) should fire within 10s";
-
-		EXPECT_EQ(g_error_resolved.load(), GOOD_CALLS) << "All valid calls should resolve";
-
-		EXPECT_GE(g_error_rejected.load(), 1) << "Bad endpoint call should be rejected";
-	}
-#endif
-
-	metacall_destroy();
-}
+}

source/tests/metacall_rpc_test/source/server.js

@@ -35,14 +35,30 @@ const server = http.createServer((req, res) => {
 	} else if (req.method === 'POST') {
 		if (req.url === '/viferga/example/v1/call/divide') {
 			data.then((body) => {
-				console.log('¡Call recieved!');
-				if (body !== '[50.0,10.0]') {
-					console.error('Invalid body:', body);
-					process.exit(1);
-				}
-				const result = '5.0';
+				console.log('Call received: divide', body);
+				const args = JSON.parse(body);
+				const result = args[0] / args[1];
+				res.setHeader('Content-Type', 'application/json');
+				const str = Number.isInteger(result) ? result.toFixed(1) : JSON.stringify(result);
+				res.end(str);
+			});
+			return;
+		} else if (req.url === '/viferga/example/v1/call/sum') {
+			data.then((body) => {
+				console.log('Call received: sum', body);
+				const args = JSON.parse(body);
+				const result = args[0] + args[1];
 				res.setHeader('Content-Type', 'application/json');
-				res.end(result);
+				res.end(JSON.stringify(result));
+			});
+			return;
+		} else if (req.url === '/viferga/example/v1/call/multiply') {
+			data.then((body) => {
+				console.log('Call received: multiply', body);
+				const args = JSON.parse(body);
+				const result = args[0] * args[1];
+				res.setHeader('Content-Type', 'application/json');
+				res.end(JSON.stringify(result));
 			});
 			return;
 		} else if (req.url === '/viferga/example/v1/await/async_divide') {
@@ -68,4 +84,4 @@ const server = http.createServer((req, res) => {
 
 server.listen(port, () => {
 	console.log(`MetaCall server listening at ${port}`);
-});
+});