重构ldbc负载均衡器，支持按外部、内部二元组、四元组分流。对应调整内部接口，支持send_option的rehash功能。

1 files changed, 39 insertions, 0 deletions

diff --git a/app/src/marsio.c b/app/src/marsio.c
index 7a5559a..fe6385c 100644
--- a/app/src/marsio.c
+++ b/app/src/marsio.c
@@ -137,6 +137,38 @@ static int mrapp_ctrlmsg_init(struct mr_instance * instance)
     return RT_SUCCESS;
 }
 
+static int mrapp_distributer_init(struct mr_instance * instance)
+{
+	unsigned int distmode = LDBC_DIST_OUTER_TUPLE2;
+	unsigned int hashmode = LDBC_HASH_SYM_CRC;
+
+	MESA_load_profile_uint_def(instance->g_cfgfile_path, "service", "distmode",
+		&distmode, LDBC_DIST_OUTER_TUPLE2);
+	MESA_load_profile_uint_def(instance->g_cfgfile_path, "service", "hashmode",
+		&hashmode, LDBC_HASH_SYM_CRC);
+
+	if (distmode < 0 || distmode >= LDBC_DIST_MAX)
+	{
+		MR_CFGERR_INVALID_FORMAT(instance->g_cfgfile_path, "service", "distmode");
+		return RT_ERR;
+	}
+
+	if (hashmode < 0 || distmode >= LDBC_HASH_MAX)
+	{
+		MR_CFGERR_INVALID_FORMAT(instance->g_cfgfile_path, "service", "hashmode");
+		return RT_ERR;
+	}
+
+	instance->dist_object = distributer_create(distmode, hashmode, 0);
+	if (instance->dist_object == NULL)
+	{
+		MR_ERROR("Create distributer handler failed. ");
+		return RT_ERR;
+	}
+
+	return RT_SUCCESS;
+}
+
 /* 注册应用 */
 static int mrapp_app_register(struct mr_instance * instance)
 {
@@ -455,6 +487,13 @@ int marsio_init(struct mr_instance * instance, const char * appsym)
         goto err;
     }
 
+	/* 负载均衡器 */
+	if (mrapp_distributer_init(instance) != RT_SUCCESS)
+	{
+		MR_ERROR("Distributer initialization failed.");
+		goto err;
+	}
+
     instance->neigh = malloc(sizeof(struct neighbour_manager));
     MR_VERIFY_MALLOC(instance->neigh);