libknet: add sync host to host messagging

Signed-off-by: Fabio M. Di Nitto <fdinitto@redhat.com>

libknet/handle.c

@@ -101,7 +101,10 @@ knet_handle_t knet_handle_new(const struct knet_handle_cfg *knet_handle_cfg)
 
 	memset(knet_h->pingbuf, 0, KNET_PINGBUFSIZE);
 
-	if (pthread_rwlock_init(&knet_h->list_rwlock, NULL) != 0) {
+	if ((pthread_rwlock_init(&knet_h->list_rwlock, NULL) != 0) ||
+	    (pthread_rwlock_init(&knet_h->host_rwlock, NULL) != 0) ||
+	    (pthread_mutex_init(&knet_h->host_mutex, NULL) != 0) ||
+	    (pthread_cond_init(&knet_h->host_cond, NULL) != 0)) {
 		log_err(knet_h, KNET_SUB_HANDLE, "Unable to initialize locks");
 		goto exit_fail5;
 	}
@@ -201,6 +204,9 @@ knet_handle_t knet_handle_new(const struct knet_handle_cfg *knet_handle_cfg)
 		close(knet_h->dst_link_handler_epollfd);
 
 	pthread_rwlock_destroy(&knet_h->list_rwlock);
+	pthread_rwlock_destroy(&knet_h->host_rwlock);
+	pthread_mutex_destroy(&knet_h->host_mutex);
+	pthread_cond_destroy(&knet_h->host_cond);
 
 exit_fail5:
 	free(knet_h->pingbuf);
@@ -287,6 +293,9 @@ int knet_handle_free(knet_handle_t knet_h)
 	close(knet_h->hostpipefd[1]);
 
 	pthread_rwlock_destroy(&knet_h->list_rwlock);
+	pthread_rwlock_destroy(&knet_h->host_rwlock);
+	pthread_mutex_destroy(&knet_h->host_mutex);
+	pthread_cond_destroy(&knet_h->host_cond);
 
 	free(knet_h->tap_to_links_buf);
 	free(knet_h->tap_to_links_buf_crypt);
@@ -599,6 +608,14 @@ static void _handle_tap_to_links(knet_handle_t knet_h, int sockfd)
 			}
 		}
 	}
+
+	if (knet_h->tap_to_links_buf->kf_type == KNET_FRAME_HOST_INFO) {
+		if (pthread_mutex_lock(&knet_h->host_mutex) != 0)
+			log_debug(knet_h, KNET_SUB_TAP_T, "Unable to get mutex lock");
+		pthread_cond_signal(&knet_h->host_cond);
+		pthread_mutex_unlock(&knet_h->host_mutex);
+	}
+
 	pthread_rwlock_unlock(&knet_h->list_rwlock);
 }
 

libknet/host.c

@@ -203,6 +203,41 @@ int knet_host_set_policy(knet_handle_t knet_h, uint16_t node_id, int policy)
 	return ret;
 }
 
+int _send_host_info(knet_handle_t knet_h, const void *data, const size_t datalen)
+{
+	size_t byte_cnt = 0;
+	int len;
+
+	if (pthread_rwlock_wrlock(&knet_h->host_rwlock) != 0) {
+		log_debug(knet_h, KNET_SUB_HOST, "Unable to get write lock");
+		return -1;
+	}
+
+	if (pthread_mutex_lock(&knet_h->host_mutex) != 0) {
+		log_debug(knet_h, KNET_SUB_HOST, "Unable to get mutex lock");
+		pthread_rwlock_unlock(&knet_h->host_rwlock);
+		return -1;
+	}
+
+	while (byte_cnt < datalen) {
+		len = write(knet_h->hostpipefd[1], &data, datalen - byte_cnt);
+		if (len <= 0) {
+			log_debug(knet_h, KNET_SUB_HOST, "Unable to write data to hostpipe");
+			pthread_mutex_unlock(&knet_h->host_mutex);
+			pthread_rwlock_unlock(&knet_h->host_rwlock);
+			return -1;
+		}
+
+		byte_cnt += len;
+	}
+
+	pthread_cond_wait(&knet_h->host_cond, &knet_h->host_mutex);
+	pthread_mutex_unlock(&knet_h->host_mutex);
+	pthread_rwlock_unlock(&knet_h->host_rwlock);
+
+	return 0;
+}
+
 /* bcast = 0 -> unicast packet | 1 -> broadcast|mcast */
 
 /* make this bcast/ucast aware */

libknet/libknet-private.h

@@ -50,6 +50,9 @@ struct knet_handle {
 	pthread_t heartbt_thread;
 	pthread_t dst_link_handler_thread;
 	pthread_rwlock_t list_rwlock;
+	pthread_rwlock_t host_rwlock;
+	pthread_mutex_t host_mutex;
+	pthread_cond_t host_cond;
 	struct crypto_instance *crypto_instance;
 	unsigned char *tap_to_links_buf_crypt;
 	unsigned char *recv_from_links_buf_crypt;
@@ -67,6 +70,7 @@ struct knet_handle {
 int _fdset_cloexec(int fd);
 int _fdset_nonblock(int fd);
 int _dst_cache_update(knet_handle_t knet_h, uint16_t node_id);
+int _send_host_info(knet_handle_t knet_h, const void *data, const size_t datalen);
 int _should_deliver(struct knet_host *host, int bcast, seq_num_t seq_num);
 void _has_been_delivered(struct knet_host *host, int bcast, seq_num_t seq_num);