knet: Add support for knet compression

Signed-off-by: Christine Caulfield <ccaulfie@redhat.com>
Reviewed-by: Jan Friesse <jfriesse@redhat.com>

exec/coroparse.c

@@ -618,6 +618,7 @@ static int main_config_parser_cb(const char *path,
 			    (strcmp(path, "totem.max_messages") == 0) ||
 			    (strcmp(path, "totem.miss_count_const") == 0) ||
 			    (strcmp(path, "totem.knet_pmtud_interval") == 0) ||
+			    (strcmp(path, "totem.knet_compression_threshold") == 0) ||
 			    (strcmp(path, "totem.netmtu") == 0)) {
 				val_type = ICMAP_VALUETYPE_UINT32;
 				if (safe_atoq(value, &val, val_type) != 0) {
@@ -626,6 +627,14 @@ static int main_config_parser_cb(const char *path,
 				icmap_set_uint32_r(config_map,path, val);
 				add_as_string = 0;
 			}
+			if (strcmp(path, "totem.knet_compression_level") == 0) {
+				val_type = ICMAP_VALUETYPE_INT32;
+				if (safe_atoq(value, &val, val_type) != 0) {
+					goto atoi_error;
+				}
+				icmap_set_int32_r(config_map, path, val);
+				add_as_string = 0;
+			}
 			if (strcmp(path, "totem.config_version") == 0) {
 				if (str_to_ull(value, &ull) != 0) {
 					goto atoi_error;

exec/totemconfig.c

@@ -89,9 +89,7 @@ static char error_string_response[512];
 
 static void add_totem_config_notification(struct totem_config *totem_config);
 
-
-/* All the volatile parameters are uint32s, luckily */
-static uint32_t *totem_get_param_by_name(struct totem_config *totem_config, const char *param_name)
+static void *totem_get_param_by_name(struct totem_config *totem_config, const char *param_name)
 {
 	if (strcmp(param_name, "totem.token") == 0)
 		return &totem_config->token_timeout;
@@ -127,6 +125,12 @@ static uint32_t *totem_get_param_by_name(struct totem_config *totem_config, cons
 		return &totem_config->miss_count_const;
 	if (strcmp(param_name, "totem.knet_pmtud_interval") == 0)
 		return &totem_config->knet_pmtud_interval;
+	if (strcmp(param_name, "totem.knet_compression_threshold") == 0)
+		return &totem_config->knet_compression_threshold;
+	if (strcmp(param_name, "totem.knet_compression_level") == 0)
+		return &totem_config->knet_compression_level;
+	if (strcmp(param_name, "totem.knet_compression_model") == 0)
+		return &totem_config->knet_compression_model;
 
 	return NULL;
 }
@@ -135,7 +139,7 @@ static uint32_t *totem_get_param_by_name(struct totem_config *totem_config, cons
  * Read key_name from icmap. If key is not found or key_name == delete_key or if allow_zero is false
  * and readed value is zero, default value is used and stored into totem_config.
  */
-static void totem_volatile_config_set_value (struct totem_config *totem_config,
+static void totem_volatile_config_set_uint32_value (struct totem_config *totem_config,
 	const char *key_name, const char *deleted_key, unsigned int default_value,
 	int allow_zero_value)
 {
@@ -143,8 +147,64 @@ static void totem_volatile_config_set_value (struct totem_config *totem_config,
 
 	if (icmap_get_uint32(key_name, totem_get_param_by_name(totem_config, key_name)) != CS_OK ||
 	    (deleted_key != NULL && strcmp(deleted_key, key_name) == 0) ||
-	    (!allow_zero_value && *totem_get_param_by_name(totem_config, key_name) == 0)) {
-		*totem_get_param_by_name(totem_config, key_name) = default_value;
+	    (!allow_zero_value && *(uint32_t *)totem_get_param_by_name(totem_config, key_name) == 0)) {
+		*(uint32_t *)totem_get_param_by_name(totem_config, key_name) = default_value;
+	}
+
+	/*
+	 * Store totem_config value to cmap runtime section
+	 */
+	if (strlen("runtime.config.") + strlen(key_name) >= ICMAP_KEYNAME_MAXLEN) {
+		/*
+		 * This shouldn't happen
+		 */
+		return ;
+	}
+
+	strcpy(runtime_key_name, "runtime.config.");
+	strcat(runtime_key_name, key_name);
+
+	icmap_set_uint32(runtime_key_name, *(uint32_t *)totem_get_param_by_name(totem_config, key_name));
+}
+
+static void totem_volatile_config_set_int32_value (struct totem_config *totem_config,
+	const char *key_name, const char *deleted_key, int default_value,
+	int allow_zero_value)
+{
+	char runtime_key_name[ICMAP_KEYNAME_MAXLEN];
+
+	if (icmap_get_int32(key_name, totem_get_param_by_name(totem_config, key_name)) != CS_OK ||
+	    (deleted_key != NULL && strcmp(deleted_key, key_name) == 0) ||
+	    (!allow_zero_value && *(int32_t *)totem_get_param_by_name(totem_config, key_name) == 0)) {
+		*(int32_t *)totem_get_param_by_name(totem_config, key_name) = default_value;
+	}
+
+	/*
+	 * Store totem_config value to cmap runtime section
+	 */
+	if (strlen("runtime.config.") + strlen(key_name) >= ICMAP_KEYNAME_MAXLEN) {
+		/*
+		 * This shouldn't happen
+		 */
+		return ;
+	}
+
+	strcpy(runtime_key_name, "runtime.config.");
+	strcat(runtime_key_name, key_name);
+
+	icmap_set_int32(runtime_key_name, *(int32_t *)totem_get_param_by_name(totem_config, key_name));
+}
+
+static void totem_volatile_config_set_string_value (struct totem_config *totem_config,
+	const char *key_name, const char *deleted_key, const char *default_value)
+{
+	char runtime_key_name[ICMAP_KEYNAME_MAXLEN];
+	const void **config_value;
+
+	config_value = totem_get_param_by_name(totem_config, key_name);
+	if (icmap_get_string(key_name, totem_get_param_by_name(totem_config, key_name)) != CS_OK ||
+	    (deleted_key != NULL && strcmp(deleted_key, key_name) == 0)) {
+		*config_value = default_value;
 	}
 
 	/*
@@ -160,7 +220,7 @@ static void totem_volatile_config_set_value (struct totem_config *totem_config,
 	strcpy(runtime_key_name, "runtime.config.");
 	strcat(runtime_key_name, key_name);
 
-	icmap_set_uint32(runtime_key_name, *totem_get_param_by_name(totem_config, key_name));
+	icmap_set_string(runtime_key_name, (char *)*config_value);
 }
 
 
@@ -173,10 +233,10 @@ static void totem_volatile_config_read (struct totem_config *totem_config, const
 {
 	uint32_t u32;
 
-	totem_volatile_config_set_value(totem_config, "totem.token_retransmits_before_loss_const", deleted_key,
+	totem_volatile_config_set_uint32_value(totem_config, "totem.token_retransmits_before_loss_const", deleted_key,
 	    TOKEN_RETRANSMITS_BEFORE_LOSS_CONST, 0);
 
-	totem_volatile_config_set_value(totem_config, "totem.token", deleted_key, TOKEN_TIMEOUT, 0);
+	totem_volatile_config_set_uint32_value(totem_config, "totem.token", deleted_key, TOKEN_TIMEOUT, 0);
 
 	if (totem_config->interfaces[0].member_count > 2) {
 		u32 = TOKEN_COEFFICIENT;
@@ -189,38 +249,46 @@ static void totem_volatile_config_read (struct totem_config *totem_config, const
 		icmap_set_uint32("runtime.config.totem.token", totem_config->token_timeout);
 	}
 
-	totem_volatile_config_set_value(totem_config, "totem.max_network_delay", deleted_key, MAX_NETWORK_DELAY, 0);
+	totem_volatile_config_set_uint32_value(totem_config, "totem.max_network_delay", deleted_key, MAX_NETWORK_DELAY, 0);
 
-	totem_volatile_config_set_value(totem_config, "totem.window_size", deleted_key, WINDOW_SIZE, 0);
+	totem_volatile_config_set_uint32_value(totem_config, "totem.window_size", deleted_key, WINDOW_SIZE, 0);
 
-	totem_volatile_config_set_value(totem_config, "totem.max_messages", deleted_key, MAX_MESSAGES, 0);
+	totem_volatile_config_set_uint32_value(totem_config, "totem.max_messages", deleted_key, MAX_MESSAGES, 0);
 
-	totem_volatile_config_set_value(totem_config, "totem.miss_count_const", deleted_key, MISS_COUNT_CONST, 0);
-	totem_volatile_config_set_value(totem_config, "totem.knet_pmtud_interval", deleted_key, KNET_PMTUD_INTERVAL, 0);
+	totem_volatile_config_set_uint32_value(totem_config, "totem.miss_count_const", deleted_key, MISS_COUNT_CONST, 0);
+	totem_volatile_config_set_uint32_value(totem_config, "totem.knet_pmtud_interval", deleted_key, KNET_PMTUD_INTERVAL, 0);
 
-	totem_volatile_config_set_value(totem_config, "totem.token_retransmit", deleted_key,
+	totem_volatile_config_set_uint32_value(totem_config, "totem.token_retransmit", deleted_key,
 	   (int)(totem_config->token_timeout / (totem_config->token_retransmits_before_loss_const + 0.2)), 0);
 
-	totem_volatile_config_set_value(totem_config, "totem.hold", deleted_key,
+	totem_volatile_config_set_uint32_value(totem_config, "totem.hold", deleted_key,
 	    (int)(totem_config->token_retransmit_timeout * 0.8 - (1000/HZ)), 0);
 
-	totem_volatile_config_set_value(totem_config, "totem.join", deleted_key, JOIN_TIMEOUT, 0);
+	totem_volatile_config_set_uint32_value(totem_config, "totem.join", deleted_key, JOIN_TIMEOUT, 0);
 
-	totem_volatile_config_set_value(totem_config, "totem.consensus", deleted_key,
+	totem_volatile_config_set_uint32_value(totem_config, "totem.consensus", deleted_key,
 	    (int)(float)(1.2 * totem_config->token_timeout), 0);
 
-	totem_volatile_config_set_value(totem_config, "totem.merge", deleted_key, MERGE_TIMEOUT, 0);
+	totem_volatile_config_set_uint32_value(totem_config, "totem.merge", deleted_key, MERGE_TIMEOUT, 0);
 
-	totem_volatile_config_set_value(totem_config, "totem.downcheck", deleted_key, DOWNCHECK_TIMEOUT, 0);
+	totem_volatile_config_set_uint32_value(totem_config, "totem.downcheck", deleted_key, DOWNCHECK_TIMEOUT, 0);
 
-	totem_volatile_config_set_value(totem_config, "totem.fail_recv_const", deleted_key, FAIL_TO_RECV_CONST, 0);
+	totem_volatile_config_set_uint32_value(totem_config, "totem.fail_recv_const", deleted_key, FAIL_TO_RECV_CONST, 0);
 
-	totem_volatile_config_set_value(totem_config, "totem.seqno_unchanged_const", deleted_key,
+	totem_volatile_config_set_uint32_value(totem_config, "totem.seqno_unchanged_const", deleted_key,
 	    SEQNO_UNCHANGED_CONST, 0);
 
-	totem_volatile_config_set_value(totem_config, "totem.send_join", deleted_key, 0, 1);
+	totem_volatile_config_set_uint32_value(totem_config, "totem.send_join", deleted_key, 0, 1);
+
+	totem_volatile_config_set_uint32_value(totem_config, "totem.heartbeat_failures_allowed", deleted_key, 0, 1);
+
+	totem_volatile_config_set_uint32_value(totem_config, "totem.knet_compression_threshold", deleted_key, 0, 1);
+
+	totem_volatile_config_set_int32_value(totem_config, "totem.knet_compression_level", deleted_key, 0, 1);
+
+	totem_volatile_config_set_string_value(totem_config, "totem.knet_compression_model", deleted_key, "none");
+
 
-	totem_volatile_config_set_value(totem_config, "totem.heartbeat_failures_allowed", deleted_key, 0, 1);
 }
 
 static int totem_volatile_config_validate (
@@ -1755,6 +1823,10 @@ static void totem_reload_notify(
 		if (local_node_pos != -1) {
 			icmap_set_uint32("nodelist.local_node_pos", local_node_pos);
 		}
+
+		/* Reconfigure network params as appropriate */
+		totempg_reconfigure();
+
 		free(totem_config->orig_interfaces);
 
 		icmap_set_uint8("config.totemconfig_reload_in_progress", 0);

exec/totemknet.c

@@ -906,6 +906,10 @@ int totemknet_initialize (
 		}
 		knet_log_printf(LOG_INFO, "kronosnet crypto initialized: %s/%s", crypto_cfg.crypto_cipher_type, crypto_cfg.crypto_hash_type);
 	}
+
+	/* Set up compression */
+	totemknet_reconfigure(instance, instance->totem_config);
+
 	knet_handle_setfwd(instance->knet_handle, 1);
 
 	instance->link_mode = KNET_LINK_POLICY_PASSIVE;
@@ -1263,6 +1267,28 @@ int totemknet_member_list_rebind_ip (
 	return (0);
 }
 
+int totemknet_reconfigure (
+	void *knet_context,
+	struct totem_config *totem_config)
+{
+	struct totemknet_instance *instance = (struct totemknet_instance *)knet_context;
+	struct knet_handle_compress_cfg compress_cfg;
+	int res = 0;
+
+	if (totem_config->knet_compression_model) {
+		strcpy(compress_cfg.compress_model, totem_config->knet_compression_model);
+		compress_cfg.compress_threshold = totem_config->knet_compression_threshold;
+		compress_cfg.compress_level = totem_config->knet_compression_level;
+
+		res = knet_handle_compress(instance->knet_handle, &compress_cfg);
+		if (res) {
+			KNET_LOGSYS_PERROR(errno, LOGSYS_LEVEL_ERROR, "knet_handle_compress failed");
+		}
+	}
+	return (res);
+}
+
+
 /* For the stats module */
 int totemknet_link_get_status (
 	knet_node_id_t node, uint8_t link_no,

exec/totemknet.h

@@ -138,4 +138,8 @@ extern int totemknet_member_set_active (
 	const struct totem_ip_address *member_ip,
 	int active);
 
+extern int totemknet_reconfigure (
+	void *knet_context,
+	struct totem_config *totem_config);
+
 #endif /* TOTEMKNET_H_DEFINED */

exec/totemnet.c

@@ -150,6 +150,10 @@ struct transport {
 		void *transport_context,
 		const struct totem_ip_address *member,
 		int active);
+
+	int (*reconfigure) (
+		void *net_context,
+		struct totem_config *totem_config);
 };
 
 struct transport transport_entries[] = {
@@ -171,7 +175,8 @@ struct transport transport_entries[] = {
 		.ifaces_get = totemudp_ifaces_get,
 		.token_target_set = totemudp_token_target_set,
 		.crypto_set = totemudp_crypto_set,
-		.recv_mcast_empty = totemudp_recv_mcast_empty
+		.recv_mcast_empty = totemudp_recv_mcast_empty,
+		.reconfigure = totemudp_reconfigure
 	},
 	{
 		.name = "UDP/IP Unicast",
@@ -194,6 +199,7 @@ struct transport transport_entries[] = {
 		.recv_mcast_empty = totemudpu_recv_mcast_empty,
 		.member_add = totemudpu_member_add,
 		.member_remove = totemudpu_member_remove,
+		.reconfigure = totemudpu_reconfigure
 	},
 	{
 		.name = "Kronosnet",
@@ -216,7 +222,7 @@ struct transport transport_entries[] = {
 		.recv_mcast_empty = totemknet_recv_mcast_empty,
 		.member_add = totemknet_member_add,
 		.member_remove = totemknet_member_remove,
-
+		.reconfigure = totemknet_reconfigure
 	}
 };
 
@@ -545,3 +551,17 @@ int totemnet_member_set_active (
 
 	return (res);
 }
+
+int totemnet_reconfigure (
+	void *net_context,
+	struct totem_config *totem_config)
+{
+	struct totemnet_instance *instance = (struct totemnet_instance *)net_context;
+	unsigned int res = 0;
+
+	res = instance->transport->reconfigure (
+			instance->transport_context,
+			totem_config);
+
+	return (res);
+}

exec/totemnet.h

@@ -116,6 +116,8 @@ extern int totemnet_finalize (void *net_context);
 
 extern int totemnet_net_mtu_adjust (void *net_context, struct totem_config *totem_config);
 
+extern int totemnet_reconfigure (void *net_context, struct totem_config *totem_config);
+
 extern const char *totemnet_iface_print (void *net_context);
 
 extern int totemnet_ifaces_get (

exec/totempg.c

@@ -1521,6 +1521,11 @@ extern int totempg_member_remove (
 	return totemsrp_member_remove (totemsrp_context, member, ring_no);
 }
 
+extern int totempg_reconfigure (void)
+{
+	return totemsrp_reconfigure (totemsrp_context, totempg_totem_config);
+}
+
 void totempg_threaded_mode_enable (void)
 {
 	totempg_threaded_mode = 1;

exec/totemsrp.c

@@ -4778,3 +4778,13 @@ void totemsrp_trans_ack (void *context)
 	instance->waiting_trans_ack = 0;
 	instance->totemsrp_waiting_trans_ack_cb_fn (0);
 }
+
+
+int totemsrp_reconfigure (void *context, struct totem_config *totem_config)
+{
+	struct totemsrp_instance *instance = (struct totemsrp_instance *)context;
+	int res;
+
+	res = totemnet_reconfigure (instance->totemnet_context, totem_config);
+	return (res);
+}

exec/totemsrp.h

@@ -146,4 +146,8 @@ void totemsrp_threaded_mode_enable (
 void totemsrp_trans_ack (
 	void *srp_context);
 
+int totemsrp_reconfigure (
+	void *context,
+	struct totem_config *totem_config);
+
 #endif /* TOTEMSRP_H_DEFINED */

exec/totemudp.c

@@ -1403,3 +1403,11 @@ int totemudp_iface_set (void *net_context,
 	/* Not supported */
 	return (-1);
 }
+
+int totemudp_reconfigure (
+	void *udp_context,
+	struct totem_config *totem_config)
+{
+	/* Not supported */
+	return (-1);
+}

exec/totemudp.h

@@ -122,4 +122,8 @@ extern int totemudp_crypto_set (
 extern int totemudp_recv_mcast_empty (
 	void *udp_context);
 
+extern int totemudp_reconfigure (
+	void *udp_context,
+	struct totem_config *totem_config);
+
 #endif /* TOTEMUDP_H_DEFINED */

exec/totemudpu.c

@@ -1201,3 +1201,11 @@ static void totemudpu_stop_merge_detect_timeout(
 	qb_loop_timer_del(instance->totemudpu_poll_handle,
 	    instance->timer_merge_detect_timeout);
 }
+
+int totemudpu_reconfigure (
+	void *udpu_context,
+	struct totem_config *totem_config)
+{
+	/* Not supported */
+	return (-1);
+}

exec/totemudpu.h

@@ -133,4 +133,8 @@ extern int totemudpu_member_remove (
 	const struct totem_ip_address *member,
 	int ring_no);
 
+extern int totemudpu_reconfigure (
+	void *udpu_context,
+	struct totem_config *totem_config);
+
 #endif /* TOTEMUDPU_H_DEFINED */

include/corosync/totem/totem.h

@@ -199,6 +199,12 @@ struct totem_config {
 
 	char *crypto_hash_type;
 
+	char *knet_compression_model;
+
+	uint32_t knet_compression_threshold;
+
+	int knet_compression_level;
+
 	totem_transport_t transport_number;
 
 	unsigned int miss_count_const;

include/corosync/totem/totempg.h

@@ -189,6 +189,8 @@ extern void totempg_threaded_mode_enable (void);
 
 extern void totempg_trans_ack (void);
 
+extern int totempg_reconfigure (void);
+
 #ifdef __cplusplus
 }
 #endif

man/corosync.conf.5

@@ -331,6 +331,27 @@ the corosync community.
 
 The default is 238 milliseconds.
 
+.TP
+knet_compression_model
+The (optional) type of compression used by kronosnet. The values available depend on
+the build and also avaialable libraries. Typically zlib and lz4 will be available
+but bzip2 and others could also be allowed. The default is 'none'
+
+.TP
+knet_compression_threshold
+Tells knet to NOT compress any packets that are smaller than the value
+indicated. Default 100 bytes.
+
+Set to 0 to reset to the default.
+Set to 1 to compress everything.
+
+.TP
+knet_compression_level
+Many compression libraries allow tuning of compression parameters. For example
+0 or 1 ... 9 are commonly used to determine the level of compression. This value
+is passed unmodified to the compression library so it is recommended to consult
+the library's documentation for more detailed information.
+
 .TP
 hold
 This timeout specifies in milliseconds how long the token should be held by