handle_api.c
No OneTemporary
Actions

Size

16 KB

Referenced Files

None

Subscribers

None

handle_api.c
View Options

	/*
	* Copyright (C) 2020-2024 Red Hat, Inc. All rights reserved.
	*
	* Authors: Fabio M. Di Nitto <fabbione@kronosnet.org>
	* Federico Simoncelli <fsimon@kronosnet.org>
	*
	* This software licensed under LGPL-2.0+
	*/

	#include "config.h"

	#include <stdlib.h>
	#include <string.h>
	#include <unistd.h>
	#include <errno.h>
	#include <pthread.h>
	#include <sys/uio.h>

	#include "internals.h"
	#include "crypto.h"
	#include "host.h"
	#include "links.h"
	#include "common.h"
	#include "transport_common.h"
	#include "logging.h"

	int knet_handle_enable_sock_notify(knet_handle_t knet_h,
	void *sock_notify_fn_private_data,
	void (*sock_notify_fn) (
	void *private_data,
	int datafd,
	int8_t channel,
	uint8_t tx_rx,
	int error,
	int errorno))
	{
	int savederrno = 0;

	if (!_is_valid_handle(knet_h)) {
	return -1;
	}

	if (!sock_notify_fn) {
	errno = EINVAL;
	return -1;
	}

	savederrno = get_global_wrlock(knet_h);
	if (savederrno) {
	log_err(knet_h, KNET_SUB_HANDLE, "Unable to get write lock: %s",
	strerror(savederrno));
	errno = savederrno;
	return -1;
	}

	knet_h->sock_notify_fn_private_data = sock_notify_fn_private_data;
	knet_h->sock_notify_fn = sock_notify_fn;
	log_debug(knet_h, KNET_SUB_HANDLE, "sock_notify_fn enabled");

	pthread_rwlock_unlock(&knet_h->global_rwlock);

	return 0;
	}

	int knet_handle_add_datafd(knet_handle_t knet_h, int datafd, int8_t channel)
	{
	int err = 0, savederrno = 0;
	int i;
	struct epoll_event ev;

	if (!_is_valid_handle(knet_h)) {
	return -1;
	}

	if (datafd == NULL) {
	errno = EINVAL;
	return -1;
	}

	if (channel == NULL) {
	errno = EINVAL;
	return -1;
	}

	if (*channel >= KNET_DATAFD_MAX) {
	errno = EINVAL;
	return -1;
	}

	savederrno = get_global_wrlock(knet_h);
	if (savederrno) {
	log_err(knet_h, KNET_SUB_HANDLE, "Unable to get write lock: %s",
	strerror(savederrno));
	errno = savederrno;
	return -1;
	}

	if (!knet_h->sock_notify_fn) {
	log_err(knet_h, KNET_SUB_HANDLE, "Adding datafd requires sock notify callback enabled!");
	savederrno = EINVAL;
	err = -1;
	goto out_unlock;
	}

	if (*datafd > 0) {
	for (i = 0; i < KNET_DATAFD_MAX; i++) {
	if ((knet_h->sockfd[i].in_use) && (knet_h->sockfd[i].sockfd[0] == *datafd)) {
	log_err(knet_h, KNET_SUB_HANDLE, "requested datafd: %d already exist in index: %d", *datafd, i);
	savederrno = EEXIST;
	err = -1;
	goto out_unlock;
	}
	}
	}

	/*
	* auto allocate a channel
	*/
	if (*channel < 0) {
	for (i = 0; i < KNET_DATAFD_MAX; i++) {
	if (!knet_h->sockfd[i].in_use) {
	*channel = i;
	break;
	}
	}
	if (*channel < 0) {
	savederrno = EBUSY;
	err = -1;
	goto out_unlock;
	}
	} else {
	if (knet_h->sockfd[*channel].in_use) {
	savederrno = EBUSY;
	err = -1;
	goto out_unlock;
	}
	}

	knet_h->sockfd[*channel].is_created = 0;
	knet_h->sockfd[*channel].is_socket = 0;
	knet_h->sockfd[*channel].has_error = 0;

	if (*datafd > 0) {
	int sockopt;
	socklen_t sockoptlen = sizeof(sockopt);

	if (_fdset_cloexec(*datafd)) {
	savederrno = errno;
	err = -1;
	log_err(knet_h, KNET_SUB_HANDLE, "Unable to set CLOEXEC on datafd: %s",
	strerror(savederrno));
	goto out_unlock;
	}

	if (_fdset_nonblock(*datafd)) {
	savederrno = errno;
	err = -1;
	log_err(knet_h, KNET_SUB_HANDLE, "Unable to set NONBLOCK on datafd: %s",
	strerror(savederrno));
	goto out_unlock;
	}

	knet_h->sockfd[channel].sockfd[0] = datafd;
	knet_h->sockfd[*channel].sockfd[1] = 0;

	if (!getsockopt(knet_h->sockfd[*channel].sockfd[0], SOL_SOCKET, SO_TYPE, &sockopt, &sockoptlen)) {
	knet_h->sockfd[*channel].is_socket = 1;
	}
	} else {
	if (_init_socketpair(knet_h, knet_h->sockfd[*channel].sockfd)) {
	savederrno = errno;
	err = -1;
	goto out_unlock;
	}

	knet_h->sockfd[*channel].is_created = 1;
	knet_h->sockfd[*channel].is_socket = 1;
	datafd = knet_h->sockfd[channel].sockfd[0];
	}

	memset(&ev, 0, sizeof(struct epoll_event));
	ev.events = EPOLLIN;
	ev.data.fd = knet_h->sockfd[channel].sockfd[knet_h->sockfd[channel].is_created];

	if (epoll_ctl(knet_h->send_to_links_epollfd,
	EPOLL_CTL_ADD, knet_h->sockfd[channel].sockfd[knet_h->sockfd[channel].is_created], &ev)) {
	savederrno = errno;
	err = -1;
	log_err(knet_h, KNET_SUB_HANDLE, "Unable to add datafd %d to linkfd epoll pool: %s",
	knet_h->sockfd[channel].sockfd[knet_h->sockfd[channel].is_created], strerror(savederrno));
	if (knet_h->sockfd[*channel].is_created) {
	_close_socketpair(knet_h, knet_h->sockfd[*channel].sockfd);
	}
	goto out_unlock;
	}

	knet_h->sockfd[*channel].in_use = 1;

	out_unlock:
	pthread_rwlock_unlock(&knet_h->global_rwlock);
	errno = err ? savederrno : 0;
	return err;
	}

	int knet_handle_remove_datafd(knet_handle_t knet_h, int datafd)
	{
	int err = 0, savederrno = 0;
	int8_t channel = -1;
	int i;
	struct epoll_event ev;

	if (!_is_valid_handle(knet_h)) {
	return -1;
	}

	if (datafd <= 0) {
	errno = EINVAL;
	return -1;
	}

	savederrno = get_global_wrlock(knet_h);
	if (savederrno) {
	log_err(knet_h, KNET_SUB_HANDLE, "Unable to get write lock: %s",
	strerror(savederrno));
	errno = savederrno;
	return -1;
	}

	for (i = 0; i < KNET_DATAFD_MAX; i++) {
	if ((knet_h->sockfd[i].in_use) &&
	(knet_h->sockfd[i].sockfd[0] == datafd)) {
	channel = i;
	break;
	}
	}

	if (channel < 0) {
	savederrno = EINVAL;
	err = -1;
	goto out_unlock;
	}

	if (!knet_h->sockfd[channel].has_error) {
	memset(&ev, 0, sizeof(struct epoll_event));

	if (epoll_ctl(knet_h->send_to_links_epollfd,
	EPOLL_CTL_DEL, knet_h->sockfd[channel].sockfd[knet_h->sockfd[channel].is_created], &ev)) {
	savederrno = errno;
	err = -1;
	log_err(knet_h, KNET_SUB_HANDLE, "Unable to del datafd %d from linkfd epoll pool: %s",
	knet_h->sockfd[channel].sockfd[0], strerror(savederrno));
	goto out_unlock;
	}
	}

	if (knet_h->sockfd[channel].is_created) {
	_close_socketpair(knet_h, knet_h->sockfd[channel].sockfd);
	}

	memset(&knet_h->sockfd[channel], 0, sizeof(struct knet_sock));

	out_unlock:
	pthread_rwlock_unlock(&knet_h->global_rwlock);
	errno = err ? savederrno : 0;
	return err;
	}

	int knet_handle_get_datafd(knet_handle_t knet_h, const int8_t channel, int *datafd)
	{
	int err = 0, savederrno = 0;

	if (!_is_valid_handle(knet_h)) {
	return -1;
	}

	if ((channel < 0) \|\| (channel >= KNET_DATAFD_MAX)) {
	errno = EINVAL;
	return -1;
	}

	if (datafd == NULL) {
	errno = EINVAL;
	return -1;
	}

	savederrno = pthread_rwlock_rdlock(&knet_h->global_rwlock);
	if (savederrno) {
	log_err(knet_h, KNET_SUB_HANDLE, "Unable to get read lock: %s",
	strerror(savederrno));
	errno = savederrno;
	return -1;
	}

	if (!knet_h->sockfd[channel].in_use) {
	savederrno = EINVAL;
	err = -1;
	goto out_unlock;
	}

	*datafd = knet_h->sockfd[channel].sockfd[0];

	out_unlock:
	pthread_rwlock_unlock(&knet_h->global_rwlock);
	errno = err ? savederrno : 0;
	return err;
	}

	int knet_handle_get_channel(knet_handle_t knet_h, const int datafd, int8_t *channel)
	{
	int err = 0, savederrno = 0;
	int i;

	if (!_is_valid_handle(knet_h)) {
	return -1;
	}

	if (datafd <= 0) {
	errno = EINVAL;
	return -1;
	}

	if (channel == NULL) {
	errno = EINVAL;
	return -1;
	}

	savederrno = pthread_rwlock_rdlock(&knet_h->global_rwlock);
	if (savederrno) {
	log_err(knet_h, KNET_SUB_HANDLE, "Unable to get read lock: %s",
	strerror(savederrno));
	errno = savederrno;
	return -1;
	}

	*channel = -1;

	for (i = 0; i < KNET_DATAFD_MAX; i++) {
	if ((knet_h->sockfd[i].in_use) &&
	(knet_h->sockfd[i].sockfd[0] == datafd)) {
	*channel = i;
	break;
	}
	}

	if (*channel < 0) {
	savederrno = EINVAL;
	err = -1;
	goto out_unlock;
	}

	out_unlock:
	pthread_rwlock_unlock(&knet_h->global_rwlock);
	errno = err ? savederrno : 0;
	return err;
	}

	int knet_handle_enable_filter(knet_handle_t knet_h,
	void *dst_host_filter_fn_private_data,
	int (*dst_host_filter_fn) (
	void *private_data,
	const unsigned char *outdata,
	ssize_t outdata_len,
	uint8_t tx_rx,
	knet_node_id_t this_host_id,
	knet_node_id_t src_node_id,
	int8_t *channel,
	knet_node_id_t *dst_host_ids,
	size_t *dst_host_ids_entries))
	{
	int savederrno = 0;

	if (!_is_valid_handle(knet_h)) {
	return -1;
	}

	savederrno = get_global_wrlock(knet_h);
	if (savederrno) {
	log_err(knet_h, KNET_SUB_HANDLE, "Unable to get write lock: %s",
	strerror(savederrno));
	errno = savederrno;
	return -1;
	}

	knet_h->dst_host_filter_fn_private_data = dst_host_filter_fn_private_data;
	knet_h->dst_host_filter_fn = dst_host_filter_fn;
	if (knet_h->dst_host_filter_fn) {
	log_debug(knet_h, KNET_SUB_HANDLE, "dst_host_filter_fn enabled");
	} else {
	log_debug(knet_h, KNET_SUB_HANDLE, "dst_host_filter_fn disabled");
	}

	pthread_rwlock_unlock(&knet_h->global_rwlock);

	errno = 0;
	return 0;
	}

	int knet_handle_setfwd(knet_handle_t knet_h, unsigned int enabled)
	{
	int savederrno = 0;

	if (!_is_valid_handle(knet_h)) {
	return -1;
	}

	if (enabled > 1) {
	errno = EINVAL;
	return -1;
	}

	savederrno = get_global_wrlock(knet_h);
	if (savederrno) {
	log_err(knet_h, KNET_SUB_HANDLE, "Unable to get write lock: %s",
	strerror(savederrno));
	errno = savederrno;
	return -1;
	}

	if (enabled) {
	knet_h->enabled = enabled;
	log_debug(knet_h, KNET_SUB_HANDLE, "Data forwarding is enabled");
	} else {
	/*
	* notify TX and RX threads to flush the queues
	*/
	if (set_thread_flush_queue(knet_h, KNET_THREAD_TX, KNET_THREAD_QUEUE_FLUSH) < 0) {
	log_debug(knet_h, KNET_SUB_HANDLE, "Unable to request queue flushing for TX thread");
	}
	if (set_thread_flush_queue(knet_h, KNET_THREAD_RX, KNET_THREAD_QUEUE_FLUSH) < 0) {
	log_debug(knet_h, KNET_SUB_HANDLE, "Unable to request queue flushing for RX thread");
	}
	}

	pthread_rwlock_unlock(&knet_h->global_rwlock);

	/*
	* when disabling data forward, we need to give time to TX and RX
	* to flush the queues.
	*
	* the TX thread is the main leader here. When there is no more
	* data in the TX queue, we will also close traffic for RX.
	*/
	if (!enabled) {
	/*
	* this usleep might be unnecessary, but wait_all_threads_flush_queue
	* adds extra locking delay.
	*
	* allow all threads to run free without extra locking interference
	* and then we switch to a more active wait in case the scheduler
	* has decided to delay one thread or another
	*/
	usleep(knet_h->threads_timer_res * 2);
	wait_all_threads_flush_queue(knet_h);

	/*
	* all threads have done flushing the queue, we can stop data forwarding
	*/
	savederrno = get_global_wrlock(knet_h);
	if (savederrno) {
	log_err(knet_h, KNET_SUB_HANDLE, "Unable to get write lock: %s",
	strerror(savederrno));
	errno = savederrno;
	return -1;
	}
	knet_h->enabled = enabled;
	log_debug(knet_h, KNET_SUB_HANDLE, "Data forwarding is disabled");
	pthread_rwlock_unlock(&knet_h->global_rwlock);
	}

	errno = 0;
	return 0;
	}

	int knet_handle_get_stats(knet_handle_t knet_h, struct knet_handle_stats *stats, size_t struct_size)
	{
	int err = 0, savederrno = 0;

	if (!_is_valid_handle(knet_h)) {
	return -1;
	}

	if (!stats) {
	errno = EINVAL;
	return -1;
	}

	savederrno = pthread_rwlock_rdlock(&knet_h->global_rwlock);
	if (savederrno) {
	log_err(knet_h, KNET_SUB_HANDLE, "Unable to get read lock: %s",
	strerror(savederrno));
	errno = savederrno;
	return -1;
	}

	savederrno = pthread_mutex_lock(&knet_h->handle_stats_mutex);
	if (savederrno) {
	log_err(knet_h, KNET_SUB_HANDLE, "Unable to get mutex lock: %s",
	strerror(savederrno));
	err = -1;
	goto out_unlock;
	}

	if (struct_size > sizeof(struct knet_handle_stats)) {
	struct_size = sizeof(struct knet_handle_stats);
	}

	memmove(stats, &knet_h->stats, struct_size);

	/*
	* TX crypt stats only count the data packets sent, so add in the ping/pong/pmtud figures
	* RX is OK as it counts them before they are sorted.
	*/

	stats->tx_crypt_packets += knet_h->stats_extra.tx_crypt_ping_packets +
	knet_h->stats_extra.tx_crypt_pong_packets +
	knet_h->stats_extra.tx_crypt_pmtu_packets +
	knet_h->stats_extra.tx_crypt_pmtu_reply_packets;

	/* Tell the caller our full size in case they have an old version */
	stats->size = sizeof(struct knet_handle_stats);

	out_unlock:
	pthread_mutex_unlock(&knet_h->handle_stats_mutex);
	pthread_rwlock_unlock(&knet_h->global_rwlock);
	return err;
	}

	int knet_handle_clear_stats(knet_handle_t knet_h, int clear_option)
	{
	int savederrno = 0;

	if (!_is_valid_handle(knet_h)) {
	return -1;
	}

	if (clear_option != KNET_CLEARSTATS_HANDLE_ONLY &&
	clear_option != KNET_CLEARSTATS_HANDLE_AND_LINK) {
	errno = EINVAL;
	return -1;
	}

	savederrno = get_global_wrlock(knet_h);
	if (savederrno) {
	log_err(knet_h, KNET_SUB_HANDLE, "Unable to get write lock: %s",
	strerror(savederrno));
	errno = savederrno;
	return -1;
	}

	memset(&knet_h->stats, 0, sizeof(struct knet_handle_stats));
	memset(&knet_h->stats_extra, 0, sizeof(struct knet_handle_stats_extra));
	if (clear_option == KNET_CLEARSTATS_HANDLE_AND_LINK) {
	_link_clear_stats(knet_h);
	}

	pthread_rwlock_unlock(&knet_h->global_rwlock);
	return 0;
	}

	int knet_handle_enable_access_lists(knet_handle_t knet_h, unsigned int enabled)
	{
	int savederrno = 0;

	if (!_is_valid_handle(knet_h)) {
	return -1;
	}

	if (enabled > 1) {
	errno = EINVAL;
	return -1;
	}

	savederrno = get_global_wrlock(knet_h);
	if (savederrno) {
	log_err(knet_h, KNET_SUB_HANDLE, "Unable to get write lock: %s",
	strerror(savederrno));
	errno = savederrno;
	return -1;
	}

	knet_h->use_access_lists = enabled;

	if (enabled) {
	log_debug(knet_h, KNET_SUB_HANDLE, "Links access lists are enabled");
	} else {
	log_debug(knet_h, KNET_SUB_HANDLE, "Links access lists are disabled");
	}

	pthread_rwlock_unlock(&knet_h->global_rwlock);

	errno = 0;
	return 0;
	}

	int knet_handle_get_host_defrag_bufs(knet_handle_t knet_h,
	uint16_t *min_defrag_bufs,
	uint16_t *max_defrag_bufs,
	uint8_t *shrink_threshold,
	defrag_bufs_reclaim_policy_t *reclaim_policy)
	{
	int savederrno = 0;

	if (!_is_valid_handle(knet_h)) {
	return -1;
	}

	if (!min_defrag_bufs) {
	errno = EINVAL;
	return -1;
	}

	if (!max_defrag_bufs) {
	errno = EINVAL;
	return -1;
	}

	if (!shrink_threshold) {
	errno = EINVAL;
	return -1;
	}

	if (!reclaim_policy) {
	errno = EINVAL;
	return -1;
	}

	savederrno = pthread_rwlock_rdlock(&knet_h->global_rwlock);
	if (savederrno) {
	log_err(knet_h, KNET_SUB_HANDLE, "Unable to get read lock: %s",
	strerror(savederrno));
	errno = savederrno;
	return -1;
	}

	*min_defrag_bufs = knet_h->defrag_bufs_min;
	*max_defrag_bufs = knet_h->defrag_bufs_max;
	*shrink_threshold = knet_h->defrag_bufs_shrink_threshold;
	*reclaim_policy = knet_h->defrag_bufs_reclaim_policy;

	pthread_rwlock_unlock(&knet_h->global_rwlock);
	errno = 0;
	return 0;
	}

	static int _is_power_of_two(uint16_t num)
	{
	if ((num != 0) && ((num &(num - 1)) == 0)) {
	return 1;
	}
	return 0;
	}

	int knet_handle_set_host_defrag_bufs(knet_handle_t knet_h,
	uint16_t min_defrag_bufs,
	uint16_t max_defrag_bufs,
	uint8_t shrink_threshold,
	defrag_bufs_reclaim_policy_t reclaim_policy)
	{
	int err = 0, savederrno = 0;
	struct knet_host *host;

	if (!_is_valid_handle(knet_h)) {
	return -1;
	}

	if ((!min_defrag_bufs) \|\|
	(!_is_power_of_two(min_defrag_bufs)) \|\|
	(min_defrag_bufs > max_defrag_bufs)) {
	errno = EINVAL;
	return -1;
	}

	if ((!max_defrag_bufs) \|\|
	(!_is_power_of_two(max_defrag_bufs)) \|\|
	(max_defrag_bufs < min_defrag_bufs)) {
	errno = EINVAL;
	return -1;
	}

	if ((!shrink_threshold) \|\|
	(shrink_threshold > 50)) {
	errno = EINVAL;
	return -1;
	}

	if (reclaim_policy > 1) {
	errno = EINVAL;
	return -1;
	}

	savederrno = get_global_wrlock(knet_h);
	if (savederrno) {
	log_err(knet_h, KNET_SUB_HANDLE, "Unable to get write lock: %s",
	strerror(savederrno));
	errno = savederrno;
	return -1;
	}

	/*
	* all those parameters are managed by thread_rx in read only context.
	* it is safe to change them here because we are in write lock.
	*/
	knet_h->defrag_bufs_min = min_defrag_bufs;
	knet_h->defrag_bufs_max = max_defrag_bufs;
	knet_h->defrag_bufs_shrink_threshold = shrink_threshold;
	knet_h->defrag_bufs_reclaim_policy = reclaim_policy;

	/*
	* reset all stats based on new values
	*/
	for (host = knet_h->host_head; host != NULL; host = host->next) {
	_clear_defrag_bufs_stats(host);
	}

	pthread_rwlock_unlock(&knet_h->global_rwlock);
	errno = 0;
	return err;
	}

File Metadata

Mime Type: text/x-c
Expires: Thu, Jun 5, 11:12 PM (4 h, 26 m)
Storage Engine: blob
Storage Format: Raw Data
Storage Handle: 1861750
Default Alt Text: handle_api.c (16 KB)

handle_api.cNo OneTemporaryActions

handle_api.cView Options

File Metadata

Event Timeline

handle_api.c
No OneTemporary
Actions

handle_api.c
View Options