libqb/lib/ringbuffer_helper.c - actifio - Git at Google

 /*
  * Copyright (C) 2010 Red Hat, Inc.
  *
  * Author: Angus Salkeld <asalkeld@redhat.com>
  *
  * This file is part of libqb.
  *
  * libqb is free software: you can redistribute it and/or modify
  * it under the terms of the GNU Lesser General Public License as published by
  * the Free Software Foundation, either version 2.1 of the License, or
  * (at your option) any later version.
  *
  * libqb is distributed in the hope that it will be useful,
  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  * GNU Lesser General Public License for more details.
  *
  * You should have received a copy of the GNU Lesser General Public License
  * along with libqb.  If not, see <http://www.gnu.org/licenses/>.
  */
 #include "ringbuffer_int.h"
 #include <qb/qbdefs.h>

 static int32_t
 my_posix_sem_timedwait(void * instance, int32_t ms_timeout)
 {
 	struct qb_ringbuffer_s *rb = (struct qb_ringbuffer_s *)instance;
 	struct timespec ts_timeout;
 	int32_t res;

 	if (ms_timeout > 0) {
 		qb_util_timespec_from_epoch_get(&ts_timeout);
 		qb_timespec_add_ms(&ts_timeout, ms_timeout);
 	}

 sem_wait_again:
 	if (ms_timeout > 0) {
 		res = rpl_sem_timedwait(&rb->shared_hdr->posix_sem, &ts_timeout);
 	} else if (ms_timeout == 0) {
 		res = rpl_sem_trywait(&rb->shared_hdr->posix_sem);
 	} else {
 		res = rpl_sem_wait(&rb->shared_hdr->posix_sem);
 	}
 	if (res == -1) {
 		switch (errno) {
 		case EINTR:
 			goto sem_wait_again;
 			break;
 		case EAGAIN:
 			res = -ETIMEDOUT;
 			break;
 		case ETIMEDOUT:
 			res = -errno;
 			break;
 		default:
 			res = -errno;
 			qb_util_perror(LOG_ERR, "error waiting for semaphore");
 			break;
 		}
 	}
 	return res;
 }

 static int32_t
 my_posix_sem_post(void * instance, size_t msg_size)
 {
 	struct qb_ringbuffer_s *rb = (struct qb_ringbuffer_s *)instance;
 	if (rpl_sem_post(&rb->shared_hdr->posix_sem) < 0) {
 		return -errno;
 	} else {
 		return 0;
 	}
 }

 static ssize_t
 my_posix_getvalue_fn(void * instance)
 {
 	struct qb_ringbuffer_s *rb = (struct qb_ringbuffer_s *)instance;
 	int val;
 	if (rpl_sem_getvalue(&rb->shared_hdr->posix_sem, &val) < 0) {
 		return -errno;
 	} else {
 		return val;
 	}
 }

 static int32_t
 my_posix_sem_destroy(void * instance)
 {
 	struct qb_ringbuffer_s *rb = (struct qb_ringbuffer_s *)instance;
 	qb_enter();
 	if (rpl_sem_destroy(&rb->shared_hdr->posix_sem) == -1) {
 		return -errno;
 	} else {
 		return 0;
 	}
 }

 static int32_t
 my_posix_sem_create(void * instance, uint32_t flags)
 {
 	struct qb_ringbuffer_s *rb = (struct qb_ringbuffer_s *)instance;
 	int32_t pshared = QB_FALSE;
 	if (flags & QB_RB_FLAG_SHARED_PROCESS) {
 		if ((flags & QB_RB_FLAG_CREATE) == 0) {
 			return 0;
 		}
 		pshared = QB_TRUE;
 	}
 	if (rpl_sem_init(&rb->shared_hdr->posix_sem, pshared, 0) == -1) {
 		return -errno;
 	} else {
 		return 0;
 	}
 }

 static int32_t
 my_sysv_sem_timedwait(void * instance, int32_t ms_timeout)
 {
 	struct qb_ringbuffer_s *rb = (struct qb_ringbuffer_s *)instance;
 	struct sembuf sops[1];
 	int32_t res = 0;
 #ifdef HAVE_SEMTIMEDOP
 	struct timespec ts_timeout;
 	struct timespec *ts_pt;

 	if (ms_timeout >= 0) {
 		/*
 		 * Note: sem_timedwait takes an absolute time where as semtimedop
 		 * takes a relative time.
 		 */
 		ts_timeout.tv_sec = 0;
 		ts_timeout.tv_nsec = 0;
 		qb_timespec_add_ms(&ts_timeout, ms_timeout);
 		ts_pt = &ts_timeout;
 	} else {
 		ts_pt = NULL;
 	}
 #endif /* HAVE_SEMTIMEDOP */

 	/*
 	 * wait for sem post.
 	 */
 	sops[0].sem_num = 0;
 	sops[0].sem_op = -1;
 #ifdef HAVE_SEMTIMEDOP
 	sops[0].sem_flg = 0;
 #else
 	sops[0].sem_flg = IPC_NOWAIT;
 #endif /* HAVE_SEMTIMEDOP */

 semop_again:
 #ifdef HAVE_SEMTIMEDOP
 	if (semtimedop(rb->sem_id, sops, 1, ts_pt) == -1)
 #else
 	if (semop(rb->sem_id, sops, 1) == -1)
 #endif /* HAVE_SEMTIMEDOP */
 	{
 		if (errno == EINTR) {
 			goto semop_again;
 		} else if (errno == EAGAIN) {
 			/* make consistent with sem_timedwait */
 			res = -ETIMEDOUT;
 		} else {
 			res = -errno;
 			qb_util_perror(LOG_ERR, "error waiting for semaphore");
 		}
 		return res;
 	}
 	return 0;
 }

 static int32_t
 my_sysv_sem_post(void * instance, size_t msg_size)
 {
 	struct qb_ringbuffer_s *rb = (struct qb_ringbuffer_s *)instance;
 	struct sembuf sops[1];

 	if ((rb->flags & QB_RB_FLAG_SHARED_PROCESS) == 0) {
 		return 0;
 	}

 	sops[0].sem_num = 0;
 	sops[0].sem_op = 1;
 	sops[0].sem_flg = 0;

 semop_again:
 	if (semop(rb->sem_id, sops, 1) == -1) {
 		if (errno == EINTR) {
 			goto semop_again;
 		} else {
 			qb_util_perror(LOG_ERR,
 				       "could not increment semaphore");
 		}

 		return -errno;
 	}
 	return 0;
 }

 static ssize_t
 my_sysv_getvalue_fn(void * instance)
 {
 	struct qb_ringbuffer_s *rb = (struct qb_ringbuffer_s *)instance;
 	ssize_t res = semctl(rb->sem_id, 0, GETVAL, 0);
 	if (res == -1) {
 		return -errno;
 	}
 	return res;
 }

 static int32_t
 my_sysv_sem_destroy(void * instance)
 {
 	struct qb_ringbuffer_s *rb = (struct qb_ringbuffer_s *)instance;
 	if (semctl(rb->sem_id, 0, IPC_RMID, 0) == -1) {
 		return -errno;
 	} else {
 		return 0;
 	}
 }

 static int32_t
 my_sysv_sem_create(void * instance, uint32_t flags)
 {
 	struct qb_ringbuffer_s *rb = (struct qb_ringbuffer_s *)instance;
 	union semun options;
 	int32_t res;
 	key_t sem_key;

 	sem_key = ftok(rb->shared_hdr->hdr_path, (rb->shared_hdr->word_size + 1));

 	if (sem_key == -1) {
 		res = -errno;
 		qb_util_perror(LOG_ERR, "couldn't get a sem id");
 		return res;
 	}

 	if (flags & QB_RB_FLAG_CREATE) {
 		rb->sem_id = semget(sem_key, 1, IPC_CREAT | IPC_EXCL | 0600);
 		if (rb->sem_id == -1) {
 			res = -errno;
 			qb_util_perror(LOG_ERR, "couldn't create a semaphore");
 			return res;
 		}
 		options.val = 0;
 		res = semctl(rb->sem_id, 0, SETVAL, options);
 	} else {
 		rb->sem_id = semget(sem_key, 0, 0600);
 		if (rb->sem_id == -1) {
 			res = -errno;
 			qb_util_perror(LOG_ERR, "couldn't get a sem id");
 			return res;
 		}
 		res = 0;
 	}
 	qb_util_log(LOG_DEBUG, "sem key:%d, id:%d, value:%d",
 		    (int)sem_key, rb->sem_id, semctl(rb->sem_id, 0, GETVAL, 0));

 	return res;
 }

 int32_t
 qb_rb_sem_create(struct qb_ringbuffer_s * rb, uint32_t flags)
 {
 	int32_t rc;
 	int32_t use_posix = QB_TRUE;

 	if ((flags & QB_RB_FLAG_SHARED_PROCESS) &&
 	    !(flags & QB_RB_FLAG_NO_SEMAPHORE)) {
 #if defined(HAVE_POSIX_PSHARED_SEMAPHORE) || \
     defined(HAVE_RPL_PSHARED_SEMAPHORE)
 		use_posix = QB_TRUE;
 #else
 	#ifdef HAVE_SYSV_PSHARED_SEMAPHORE
 		use_posix = QB_FALSE;
 	#else
 		return -ENOTSUP;
 	#endif /* HAVE_SYSV_PSHARED_SEMAPHORE */
 #endif /* HAVE_POSIX_PSHARED_SEMAPHORE */
 	}
 	if (flags & QB_RB_FLAG_NO_SEMAPHORE) {
 		rc = 0;
 		rb->notifier.instance = NULL;
 		rb->notifier.timedwait_fn = NULL;
 		rb->notifier.post_fn = NULL;
 		rb->notifier.q_len_fn = NULL;
 		rb->notifier.space_used_fn = NULL;
 		rb->notifier.destroy_fn = NULL;
 	} else if (use_posix) {
 		rc = my_posix_sem_create(rb, flags);
 		rb->notifier.instance = rb;
 		rb->notifier.timedwait_fn = my_posix_sem_timedwait;
 		rb->notifier.post_fn = my_posix_sem_post;
 		rb->notifier.q_len_fn = my_posix_getvalue_fn;
 		rb->notifier.space_used_fn = NULL;
 		rb->notifier.destroy_fn = my_posix_sem_destroy;
 	} else {
 		rc = my_sysv_sem_create(rb, flags);
 		rb->notifier.instance = rb;
 		rb->notifier.timedwait_fn = my_sysv_sem_timedwait;
 		rb->notifier.post_fn = my_sysv_sem_post;
 		rb->notifier.q_len_fn = my_sysv_getvalue_fn;
 		rb->notifier.space_used_fn = NULL;
 		rb->notifier.destroy_fn = my_sysv_sem_destroy;
 	}
 	return rc;
 }


 /* For qb_rb_close_helper, we need to open directory in read-only
    mode and with as lightweight + strict flags as available at
    given platform (O_PATH for the former, O_DIRECTORY for the
    latter); end result is available as RB_DIR_RO_FLAGS.
  */
 #if defined(HAVE_OPENAT) && defined(HAVE_UNLINKAT)
 #  ifndef O_DIRECTORY
 #    define RB_DIR_RO_FLAGS1 O_RDONLY
 #  else
 #    define RB_DIR_RO_FLAGS1 O_RDONLY|O_DIRECTORY
 #  endif
 #  ifndef O_PATH
 #    define RB_DIR_RO_FLAGS RB_DIR_RO_FLAGS1
 #  else
 #    define RB_DIR_RO_FLAGS RB_DIR_RO_FLAGS1|O_PATH
 #  endif

 int32_t
 qb_rb_close_helper(struct qb_ringbuffer_s * rb, int32_t unlink_it,
 		   int32_t truncate_fallback)
 {
 	int32_t res = 0, res2 = 0;
 	uint32_t word_size = rb->shared_hdr->word_size;
 	char *hdr_path = rb->shared_hdr->hdr_path;

 	if (unlink_it) {
 		qb_util_log(LOG_DEBUG, "Free'ing ringbuffer: %s", hdr_path);
 		if (rb->notifier.destroy_fn) {
 			(void)rb->notifier.destroy_fn(rb->notifier.instance);
 		}
 	} else {
 		qb_util_log(LOG_DEBUG, "Closing ringbuffer: %s", hdr_path);
 		hdr_path = NULL;
 	}

 	if (unlink_it) {
 		char *data_path = rb->shared_hdr->data_path;
 		char *sep = strrchr(data_path, '/');
 		/* we could modify data_path in-situ, but that would segfault if
 		   we hadn't write permissions to the underlying mmap'd file */
 		char dir_path[PATH_MAX];
 		int dirfd;

 		if (sep != NULL) {
 			strncpy(dir_path, data_path, sep - data_path);
 			dir_path[sep - data_path] = '\0';
 			if ((dirfd = open(dir_path, RB_DIR_RO_FLAGS)) != -1) {
 				res = qb_sys_unlink_or_truncate_at(dirfd, sep + 1,
 								   truncate_fallback);

 				/* the dirname part is assumed to be the same */
 				assert(!strncmp(dir_path, hdr_path, sep - data_path));

 				sep = hdr_path + (sep - data_path);
 				/* now, don't touch neither data_path nor hdr_path */
 				res2 = qb_sys_unlink_or_truncate_at(dirfd, sep + 1,
 								    truncate_fallback);
 				close(dirfd);
 			} else {
 				res = -errno;
 				qb_util_perror(LOG_DEBUG,
 					       "Cannot open dir: %s", hdr_path);
 			}
 		} else {
 			res = -EINVAL;
 			qb_util_perror(LOG_DEBUG,
 				       "Not dir-separable path: %s", hdr_path);
 		}
 #else
 		res = qb_sys_unlink_or_truncate(data_path, truncate_fallback);
 		res2 = qb_sys_unlink_or_truncate(hdr_path, truncate_fallback);
 #endif  /* defined(HAVE_OPENAT) && defined(HAVE_UNLINKAT) */

 		res = res ? res : res2;
 		hdr_path = NULL;
 	}  /* if (unlink_it) */

 	if (munmap(rb->shared_data, (word_size * sizeof(uint32_t)) << 1) == -1) {
 		res = res ? res : -errno;
 		qb_util_perror(LOG_DEBUG, "Cannot munmap shared_data");
 	}
 	if (munmap(rb->shared_hdr, sizeof(struct qb_ringbuffer_shared_s)) == -1) {
 		res = res ? res : -errno;
 		qb_util_perror(LOG_DEBUG, "Cannot munmap shared_hdr");
 	}
 	free(rb);
 	return res;
 }
	/*
	* Copyright (C) 2010 Red Hat, Inc.
	*
	* Author: Angus Salkeld <asalkeld@redhat.com>
	*
	* This file is part of libqb.
	*
	* libqb is free software: you can redistribute it and/or modify
	* it under the terms of the GNU Lesser General Public License as published by
	* the Free Software Foundation, either version 2.1 of the License, or
	* (at your option) any later version.
	*
	* libqb is distributed in the hope that it will be useful,
	* but WITHOUT ANY WARRANTY; without even the implied warranty of
	* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
	* GNU Lesser General Public License for more details.
	*
	* You should have received a copy of the GNU Lesser General Public License
	* along with libqb. If not, see <http://www.gnu.org/licenses/>.
	*/
	#include "ringbuffer_int.h"
	#include <qb/qbdefs.h>

	static int32_t
	my_posix_sem_timedwait(void * instance, int32_t ms_timeout)
	{
	struct qb_ringbuffer_s rb = (struct qb_ringbuffer_s )instance;
	struct timespec ts_timeout;
	int32_t res;

	if (ms_timeout > 0) {
	qb_util_timespec_from_epoch_get(&ts_timeout);
	qb_timespec_add_ms(&ts_timeout, ms_timeout);
	}

	sem_wait_again:
	if (ms_timeout > 0) {
	res = rpl_sem_timedwait(&rb->shared_hdr->posix_sem, &ts_timeout);
	} else if (ms_timeout == 0) {
	res = rpl_sem_trywait(&rb->shared_hdr->posix_sem);
	} else {
	res = rpl_sem_wait(&rb->shared_hdr->posix_sem);
	}
	if (res == -1) {
	switch (errno) {
	case EINTR:
	goto sem_wait_again;
	break;
	case EAGAIN:
	res = -ETIMEDOUT;
	break;
	case ETIMEDOUT:
	res = -errno;
	break;
	default:
	res = -errno;
	qb_util_perror(LOG_ERR, "error waiting for semaphore");
	break;
	}
	}
	return res;
	}

	static int32_t
	my_posix_sem_post(void * instance, size_t msg_size)
	{
	struct qb_ringbuffer_s rb = (struct qb_ringbuffer_s )instance;
	if (rpl_sem_post(&rb->shared_hdr->posix_sem) < 0) {
	return -errno;
	} else {
	return 0;
	}
	}

	static ssize_t
	my_posix_getvalue_fn(void * instance)
	{
	struct qb_ringbuffer_s rb = (struct qb_ringbuffer_s )instance;
	int val;
	if (rpl_sem_getvalue(&rb->shared_hdr->posix_sem, &val) < 0) {
	return -errno;
	} else {
	return val;
	}
	}

	static int32_t
	my_posix_sem_destroy(void * instance)
	{
	struct qb_ringbuffer_s rb = (struct qb_ringbuffer_s )instance;
	qb_enter();
	if (rpl_sem_destroy(&rb->shared_hdr->posix_sem) == -1) {
	return -errno;
	} else {
	return 0;
	}
	}

	static int32_t
	my_posix_sem_create(void * instance, uint32_t flags)
	{
	struct qb_ringbuffer_s rb = (struct qb_ringbuffer_s )instance;
	int32_t pshared = QB_FALSE;
	if (flags & QB_RB_FLAG_SHARED_PROCESS) {
	if ((flags & QB_RB_FLAG_CREATE) == 0) {
	return 0;
	}
	pshared = QB_TRUE;
	}
	if (rpl_sem_init(&rb->shared_hdr->posix_sem, pshared, 0) == -1) {
	return -errno;
	} else {
	return 0;
	}
	}

	static int32_t
	my_sysv_sem_timedwait(void * instance, int32_t ms_timeout)
	{
	struct qb_ringbuffer_s rb = (struct qb_ringbuffer_s )instance;
	struct sembuf sops[1];
	int32_t res = 0;
	#ifdef HAVE_SEMTIMEDOP
	struct timespec ts_timeout;
	struct timespec *ts_pt;

	if (ms_timeout >= 0) {
	/*
	* Note: sem_timedwait takes an absolute time where as semtimedop
	* takes a relative time.
	*/
	ts_timeout.tv_sec = 0;
	ts_timeout.tv_nsec = 0;
	qb_timespec_add_ms(&ts_timeout, ms_timeout);
	ts_pt = &ts_timeout;
	} else {
	ts_pt = NULL;
	}
	#endif /* HAVE_SEMTIMEDOP */

	/*
	* wait for sem post.
	*/
	sops[0].sem_num = 0;
	sops[0].sem_op = -1;
	#ifdef HAVE_SEMTIMEDOP
	sops[0].sem_flg = 0;
	#else
	sops[0].sem_flg = IPC_NOWAIT;
	#endif /* HAVE_SEMTIMEDOP */

	semop_again:
	#ifdef HAVE_SEMTIMEDOP
	if (semtimedop(rb->sem_id, sops, 1, ts_pt) == -1)
	#else
	if (semop(rb->sem_id, sops, 1) == -1)
	#endif /* HAVE_SEMTIMEDOP */
	{
	if (errno == EINTR) {
	goto semop_again;
	} else if (errno == EAGAIN) {
	/* make consistent with sem_timedwait */
	res = -ETIMEDOUT;
	} else {
	res = -errno;
	qb_util_perror(LOG_ERR, "error waiting for semaphore");
	}
	return res;
	}
	return 0;
	}

	static int32_t
	my_sysv_sem_post(void * instance, size_t msg_size)
	{
	struct qb_ringbuffer_s rb = (struct qb_ringbuffer_s )instance;
	struct sembuf sops[1];

	if ((rb->flags & QB_RB_FLAG_SHARED_PROCESS) == 0) {
	return 0;
	}

	sops[0].sem_num = 0;
	sops[0].sem_op = 1;
	sops[0].sem_flg = 0;

	semop_again:
	if (semop(rb->sem_id, sops, 1) == -1) {
	if (errno == EINTR) {
	goto semop_again;
	} else {
	qb_util_perror(LOG_ERR,
	"could not increment semaphore");
	}

	return -errno;
	}
	return 0;
	}

	static ssize_t
	my_sysv_getvalue_fn(void * instance)
	{
	struct qb_ringbuffer_s rb = (struct qb_ringbuffer_s )instance;
	ssize_t res = semctl(rb->sem_id, 0, GETVAL, 0);
	if (res == -1) {
	return -errno;
	}
	return res;
	}

	static int32_t
	my_sysv_sem_destroy(void * instance)
	{
	struct qb_ringbuffer_s rb = (struct qb_ringbuffer_s )instance;
	if (semctl(rb->sem_id, 0, IPC_RMID, 0) == -1) {
	return -errno;
	} else {
	return 0;
	}
	}

	static int32_t
	my_sysv_sem_create(void * instance, uint32_t flags)
	{
	struct qb_ringbuffer_s rb = (struct qb_ringbuffer_s )instance;
	union semun options;
	int32_t res;
	key_t sem_key;

	sem_key = ftok(rb->shared_hdr->hdr_path, (rb->shared_hdr->word_size + 1));

	if (sem_key == -1) {
	res = -errno;
	qb_util_perror(LOG_ERR, "couldn't get a sem id");
	return res;
	}

	if (flags & QB_RB_FLAG_CREATE) {
	rb->sem_id = semget(sem_key, 1, IPC_CREAT \| IPC_EXCL \| 0600);
	if (rb->sem_id == -1) {
	res = -errno;
	qb_util_perror(LOG_ERR, "couldn't create a semaphore");
	return res;
	}
	options.val = 0;
	res = semctl(rb->sem_id, 0, SETVAL, options);
	} else {
	rb->sem_id = semget(sem_key, 0, 0600);
	if (rb->sem_id == -1) {
	res = -errno;
	qb_util_perror(LOG_ERR, "couldn't get a sem id");
	return res;
	}
	res = 0;
	}
	qb_util_log(LOG_DEBUG, "sem key:%d, id:%d, value:%d",
	(int)sem_key, rb->sem_id, semctl(rb->sem_id, 0, GETVAL, 0));

	return res;
	}

	int32_t
	qb_rb_sem_create(struct qb_ringbuffer_s * rb, uint32_t flags)
	{
	int32_t rc;
	int32_t use_posix = QB_TRUE;

	if ((flags & QB_RB_FLAG_SHARED_PROCESS) &&
	!(flags & QB_RB_FLAG_NO_SEMAPHORE)) {
	#if defined(HAVE_POSIX_PSHARED_SEMAPHORE) \|\| \
	defined(HAVE_RPL_PSHARED_SEMAPHORE)
	use_posix = QB_TRUE;
	#else
	#ifdef HAVE_SYSV_PSHARED_SEMAPHORE
	use_posix = QB_FALSE;
	#else
	return -ENOTSUP;
	#endif /* HAVE_SYSV_PSHARED_SEMAPHORE */
	#endif /* HAVE_POSIX_PSHARED_SEMAPHORE */
	}
	if (flags & QB_RB_FLAG_NO_SEMAPHORE) {
	rc = 0;
	rb->notifier.instance = NULL;
	rb->notifier.timedwait_fn = NULL;
	rb->notifier.post_fn = NULL;
	rb->notifier.q_len_fn = NULL;
	rb->notifier.space_used_fn = NULL;
	rb->notifier.destroy_fn = NULL;
	} else if (use_posix) {
	rc = my_posix_sem_create(rb, flags);
	rb->notifier.instance = rb;
	rb->notifier.timedwait_fn = my_posix_sem_timedwait;
	rb->notifier.post_fn = my_posix_sem_post;
	rb->notifier.q_len_fn = my_posix_getvalue_fn;
	rb->notifier.space_used_fn = NULL;
	rb->notifier.destroy_fn = my_posix_sem_destroy;
	} else {
	rc = my_sysv_sem_create(rb, flags);
	rb->notifier.instance = rb;
	rb->notifier.timedwait_fn = my_sysv_sem_timedwait;
	rb->notifier.post_fn = my_sysv_sem_post;
	rb->notifier.q_len_fn = my_sysv_getvalue_fn;
	rb->notifier.space_used_fn = NULL;
	rb->notifier.destroy_fn = my_sysv_sem_destroy;
	}
	return rc;
	}


	/* For qb_rb_close_helper, we need to open directory in read-only
	mode and with as lightweight + strict flags as available at
	given platform (O_PATH for the former, O_DIRECTORY for the
	latter); end result is available as RB_DIR_RO_FLAGS.
	*/
	#if defined(HAVE_OPENAT) && defined(HAVE_UNLINKAT)
	# ifndef O_DIRECTORY
	# define RB_DIR_RO_FLAGS1 O_RDONLY
	# else
	# define RB_DIR_RO_FLAGS1 O_RDONLY\|O_DIRECTORY
	# endif
	# ifndef O_PATH
	# define RB_DIR_RO_FLAGS RB_DIR_RO_FLAGS1
	# else
	# define RB_DIR_RO_FLAGS RB_DIR_RO_FLAGS1\|O_PATH
	# endif

	int32_t
	qb_rb_close_helper(struct qb_ringbuffer_s * rb, int32_t unlink_it,
	int32_t truncate_fallback)
	{
	int32_t res = 0, res2 = 0;
	uint32_t word_size = rb->shared_hdr->word_size;
	char *hdr_path = rb->shared_hdr->hdr_path;

	if (unlink_it) {
	qb_util_log(LOG_DEBUG, "Free'ing ringbuffer: %s", hdr_path);
	if (rb->notifier.destroy_fn) {
	(void)rb->notifier.destroy_fn(rb->notifier.instance);
	}
	} else {
	qb_util_log(LOG_DEBUG, "Closing ringbuffer: %s", hdr_path);
	hdr_path = NULL;
	}

	if (unlink_it) {
	char *data_path = rb->shared_hdr->data_path;
	char *sep = strrchr(data_path, '/');
	/* we could modify data_path in-situ, but that would segfault if
	we hadn't write permissions to the underlying mmap'd file */
	char dir_path[PATH_MAX];
	int dirfd;

	if (sep != NULL) {
	strncpy(dir_path, data_path, sep - data_path);
	dir_path[sep - data_path] = '\0';
	if ((dirfd = open(dir_path, RB_DIR_RO_FLAGS)) != -1) {
	res = qb_sys_unlink_or_truncate_at(dirfd, sep + 1,
	truncate_fallback);

	/* the dirname part is assumed to be the same */
	assert(!strncmp(dir_path, hdr_path, sep - data_path));

	sep = hdr_path + (sep - data_path);
	/* now, don't touch neither data_path nor hdr_path */
	res2 = qb_sys_unlink_or_truncate_at(dirfd, sep + 1,
	truncate_fallback);
	close(dirfd);
	} else {
	res = -errno;
	qb_util_perror(LOG_DEBUG,
	"Cannot open dir: %s", hdr_path);
	}
	} else {
	res = -EINVAL;
	qb_util_perror(LOG_DEBUG,
	"Not dir-separable path: %s", hdr_path);
	}
	#else
	res = qb_sys_unlink_or_truncate(data_path, truncate_fallback);
	res2 = qb_sys_unlink_or_truncate(hdr_path, truncate_fallback);
	#endif /* defined(HAVE_OPENAT) && defined(HAVE_UNLINKAT) */

	res = res ? res : res2;
	hdr_path = NULL;
	} /* if (unlink_it) */

	if (munmap(rb->shared_data, (word_size * sizeof(uint32_t)) << 1) == -1) {
	res = res ? res : -errno;
	qb_util_perror(LOG_DEBUG, "Cannot munmap shared_data");
	}
	if (munmap(rb->shared_hdr, sizeof(struct qb_ringbuffer_shared_s)) == -1) {
	res = res ? res : -errno;
	qb_util_perror(LOG_DEBUG, "Cannot munmap shared_hdr");
	}
	free(rb);
	return res;
	}