/*
 * This file and its contents are supplied under the terms of the
 * Common Development and Distribution License ("CDDL"), version 1.0.
 * You may only use this file in accordance with the terms of version
 * 1.0 of the CDDL.
 *
 * A full copy of the text of the CDDL should have accompanied this
 * source.  A copy of the CDDL is also available via the Internet at
 * http://www.illumos.org/license/CDDL.
 */

/*
 * Copyright 2015 Joyent, Inc.
 */

/*
 * Merge queue
 *
 * A multi-threaded merging queue.
 *
 * The general constraint of the merge queue is that if a set of items are
 * inserted into the queue in the same order, then no matter how many threads
 * are on the scene, we will always process the items in the same order. The
 * secondary constraint is that to support environments that must be
 * single-threaded, we explicitly *must not* create a thread in the case where
 * the number of requested threads is just one.
 *
 * To that end, we've designed our queue as a circular buffer. We will grow that
 * buffer to contain enough space for all the input items, after which we'll
 * then treat it as a circular buffer.
 *
 * Items will be issued to a processing function two at a time, until there is
 * only one item remaining in the queue, at which point we will be doing any
 * merging work.
 *
 * A given queue has three different entries that we care about tracking:
 *
 * o mq_nproc   - What is the slot of the next item to process for something
 *                looking for work.
 *
 * o mq_next    - What is the slot of the next item that should be inserted into
 *                the queue.
 *
 * o mq_ncommit - What is the slot of the next item that should be committed.
 *
 * When a thread comes and looks for work, we pop entries off of the queue based
 * on the index provided by mq_nproc. At the same time, it also gets the slot
 * that it should place the result in, which is mq_next. However, because we
 * have multiple threads that are operating on the system, we want to make sure
 * that we push things onto the queue in order. We do that by allocating a slot
 * to each task and when it completes, it waits for its slot to be ready based
 * on it being the value of mq_ncommit.
 *
 * In addition, we keep track of the number of items in the queue as well as the
 * number of active workers. There's also a generation count that is used to
 * figure out when the various values might lap one another.
 *
 * The following images show what happens when we have a queue with six items
 * and whose capacity has been shrunk to six, to better fit in the screen.
 *
 *
 * 1) This is the initial configuration of the queue right before any processing
 * is done in the context of mergeq_merge(). Every box has an initial item for
 * merging in it (represented by an 'x'). Here, the mq_nproc, mq_next, and
 * mq_ncommit will all point at the initial entry. However, the mq_next has
 * already lapped around the array and thus has a generation count of one.
 *
 * The '+' characters indicate which bucket the corresponding value of mq_nproc,
 * mq_ncommit, and mq_nproc.
 *
 *                     +---++---++---++---++---++---+
 *                     | X || X || X || X || X || X |
 *                     +---++---++---++---++---++---+
 *        mq_next (g1)   +
 *     mq_ncommit (g0)   +
 *       mq_nproc (g0)   +
 *
 * 2) This shows the state right as the first thread begins to process an entry.
 * Note in this example we will have two threads processing this queue. Note,
 * mq_ncommit has not advanced. This is because the first thread has started
 * processing entries, but it has not finished, and thus we can't commit it.
 * We've incremented mq_next by one because it has gone ahead and assigned a
 * single entry. We've incremented mq_nproc by two, because we have removed two
 * entries and thus will have another set available.
 *
 *                     +---++---++---++---++---++---+     t1 - slot 0
 *                     |   ||   || X || X || X || X |     t2 - idle
 *                     +---++---++---++---++---++---+
 *        mq_next (g1)        +
 *     mq_ncommit (g0)   +
 *       mq_nproc (g0)             +
 *
 *
 * 3) This shows the state right after the second thread begins to process an
 * entry, note that the first thread has not finished. The changes are very
 * similar to the previous state, we've advanced, mq_nproc and mq_next, but not
 * mq_ncommit.
 *
 *                     +---++---++---++---++---++---+     t1 - slot 0
 *                     |   ||   ||   ||   || X || X |     t2 - slot 1
 *                     +---++---++---++---++---++---+
 *        mq_next (g1)             +
 *     mq_ncommit (g0)   +
 *       mq_nproc (g0)                       +
 *
 * 4) This shows the state after thread one has finished processing an item, but
 * before it does anything else. Note that even if thread two finishes early, it
 * cannot commit its item until thread one finishes. Here 'Y' refers to the
 * result of merging the first two 'X's.
 *
 *                     +---++---++---++---++---++---+     t1 - idle
 *                     | Y ||   ||   ||   || X || X |     t2 - slot 1
 *                     +---++---++---++---++---++---+
 *        mq_next (g1)             +
 *     mq_ncommit (g0)        +
 *       mq_nproc (g0)                       +
 *
 * 5) This shows the state after thread one has begun to process the next round
 * and after thread two has committed, but before it begins processing the next
 * item. Note that mq_nproc has wrapped around and we've bumped its generation
 * counter.
 *
 *                     +---++---++---++---++---++---+     t1 - slot 2
 *                     | Y || Y ||   ||   ||   ||   |     t2 - idle
 *                     +---++---++---++---++---++---+
 *        mq_next (g1)                  +
 *     mq_ncommit (g0)             +
 *       mq_nproc (g0)   +
 *
 * 6) Here, thread two, will take the next two Y values and thread 1 will commit
 * its 'Y'. Thread one now must wait until thread two finishes such that it can
 * do additional work.
 *
 *                     +---++---++---++---++---++---+     t1 - waiting
 *                     |   ||   || Y ||   ||   ||   |     t2 - slot 3
 *                     +---++---++---++---++---++---+
 *        mq_next (g1)                       +
 *     mq_ncommit (g0)                  +
 *       mq_nproc (g0)             +
 *
 * 7) Here, thread two has committed and thread one is about to go process the
 * final entry. The character 'Z' represents the results of merging two 'Y's.
 *
 *                     +---++---++---++---++---++---+     t1 - idle
 *                     |   ||   || Y || Z ||   ||   |     t2 - idle
 *                     +---++---++---++---++---++---+
 *        mq_next (g1)                       +
 *     mq_ncommit (g0)                       +
 *       mq_nproc (g0)             +
 *
 * 8) Here, thread one is processing the final item. Thread two is waiting in
 * mergeq_pop() for enough items to be available. In this case, it will never
 * happen; however, once all threads have finished it will break out.
 *
 *                     +---++---++---++---++---++---+     t1 - slot 4
 *                     |   ||   ||   ||   ||   ||   |     t2 - idle
 *                     +---++---++---++---++---++---+
 *        mq_next (g1)                            +
 *     mq_ncommit (g0)                       +
 *       mq_nproc (g0)                       +
 *
 * 9) This is the final state of the queue, it has a single '*' item which is
 * the final merge result. At this point, both thread one and thread two would
 * stop processing and we'll return the result to the user.
 *
 *                     +---++---++---++---++---++---+     t1 - slot 4
 *                     |   ||   ||   ||   || * ||   |     t2 - idle
 *                     +---++---++---++---++---++---+
 *        mq_next (g1)                            +
 *     mq_ncommit (g0)                       +
 *       mq_nproc (g0)                       +
 *
 *
 * Note, that if at any point in time the processing function fails, then all
 * the merges will quiesce and that error will be propagated back to the user.
 */

#include <strings.h>
#include <sys/debug.h>
#include <thread.h>
#include <synch.h>
#include <errno.h>
#include <limits.h>
#include <stdlib.h>

#include "mergeq.h"

struct mergeq {
	mutex_t mq_lock;	/* Protects items below */
	cond_t	mq_cond;	/* Condition variable */
	void **mq_items;	/* Array of items to process */
	size_t mq_nitems;	/* Number of items in the queue */
	size_t mq_cap;		/* Capacity of the items */
	size_t mq_next;		/* Place to put next entry */
	size_t mq_gnext;	/* Generation for next */
	size_t mq_nproc;	/* Index of next thing to process */
	size_t mq_gnproc;	/* Generation for next proc */
	size_t mq_ncommit;	/* Index of the next thing to commit */
	size_t mq_gncommit;	/* Commit generation */
	uint_t mq_nactthrs;	/* Number of active threads */
	uint_t mq_ndthreads;	/* Desired number of threads */
	thread_t *mq_thrs;	/* Actual threads */
	mergeq_proc_f *mq_func;	/* Processing function */
	void *mq_arg;		/* Argument for processing */
	boolean_t mq_working;	/* Are we working on processing */
	boolean_t mq_iserror;	/* Have we encountered an error? */
	int mq_error;
};

#define	MERGEQ_DEFAULT_CAP	64

static int
mergeq_error(int err)
{
	errno = err;
	return (MERGEQ_ERROR);
}

void
mergeq_fini(mergeq_t *mqp)
{
	if (mqp == NULL)
		return;

	VERIFY(mqp->mq_working != B_TRUE);

	if (mqp->mq_items != NULL)
		mergeq_free(mqp->mq_items, sizeof (void *) * mqp->mq_cap);
	if (mqp->mq_ndthreads > 0) {
		mergeq_free(mqp->mq_thrs, sizeof (thread_t) *
		    mqp->mq_ndthreads);
	}
	VERIFY0(cond_destroy(&mqp->mq_cond));
	VERIFY0(mutex_destroy(&mqp->mq_lock));
	mergeq_free(mqp, sizeof (mergeq_t));
}

int
mergeq_init(mergeq_t **outp, uint_t nthrs)
{
	int ret;
	mergeq_t *mqp;

	mqp = mergeq_alloc(sizeof (mergeq_t));
	if (mqp == NULL)
		return (mergeq_error(ENOMEM));

	bzero(mqp, sizeof (mergeq_t));
	mqp->mq_items = mergeq_alloc(sizeof (void *) * MERGEQ_DEFAULT_CAP);
	if (mqp->mq_items == NULL) {
		mergeq_free(mqp, sizeof (mergeq_t));
		return (mergeq_error(ENOMEM));
	}
	bzero(mqp->mq_items, sizeof (void *) * MERGEQ_DEFAULT_CAP);

	mqp->mq_ndthreads = nthrs - 1;
	if (mqp->mq_ndthreads > 0) {
		mqp->mq_thrs = mergeq_alloc(sizeof (thread_t) *
		    mqp->mq_ndthreads);
		if (mqp->mq_thrs == NULL) {
			mergeq_free(mqp->mq_items, sizeof (void *) *
			    MERGEQ_DEFAULT_CAP);
			mergeq_free(mqp, sizeof (mergeq_t));
			return (mergeq_error(ENOMEM));
		}
	}

	if ((ret = mutex_init(&mqp->mq_lock, USYNC_THREAD | LOCK_ERRORCHECK,
	    NULL)) != 0) {
		if (mqp->mq_ndthreads > 0) {
			mergeq_free(mqp->mq_thrs,
			    sizeof (thread_t) * mqp->mq_ndthreads);
		}
		mergeq_free(mqp->mq_items, sizeof (void *) *
		    MERGEQ_DEFAULT_CAP);
		mergeq_free(mqp, sizeof (mergeq_t));
		return (mergeq_error(ret));
	}

	if ((ret = cond_init(&mqp->mq_cond, USYNC_THREAD, NULL)) != 0) {
		VERIFY0(mutex_destroy(&mqp->mq_lock));
		if (mqp->mq_ndthreads > 0) {
			mergeq_free(mqp->mq_thrs,
			    sizeof (thread_t) * mqp->mq_ndthreads);
		}
		mergeq_free(mqp->mq_items, sizeof (void *) *
		    MERGEQ_DEFAULT_CAP);
		mergeq_free(mqp, sizeof (mergeq_t));
		return (mergeq_error(ret));
	}

	mqp->mq_cap = MERGEQ_DEFAULT_CAP;
	*outp = mqp;
	return (0);
}

static void
mergeq_reset(mergeq_t *mqp)
{
	VERIFY(MUTEX_HELD(&mqp->mq_lock));
	VERIFY(mqp->mq_working == B_FALSE);
	if (mqp->mq_cap != 0)
		bzero(mqp->mq_items, sizeof (void *) * mqp->mq_cap);
	mqp->mq_nitems = 0;
	mqp->mq_next = 0;
	mqp->mq_gnext = 0;
	mqp->mq_nproc = 0;
	mqp->mq_gnproc = 0;
	mqp->mq_ncommit = 0;
	mqp->mq_gncommit = 0;
	mqp->mq_func = NULL;
	mqp->mq_arg = NULL;
	mqp->mq_iserror = B_FALSE;
	mqp->mq_error = 0;
}

static int
mergeq_grow(mergeq_t *mqp)
{
	size_t ncap;
	void **items;

	VERIFY(MUTEX_HELD(&mqp->mq_lock));
	VERIFY(mqp->mq_working == B_FALSE);

	if (SIZE_MAX - mqp->mq_cap < MERGEQ_DEFAULT_CAP)
		return (ENOSPC);

	ncap = mqp->mq_cap + MERGEQ_DEFAULT_CAP;
	items = mergeq_alloc(ncap * sizeof (void *));
	if (items == NULL)
		return (ENOMEM);

	bzero(items, ncap * sizeof (void *));
	bcopy(mqp->mq_items, items, mqp->mq_cap * sizeof (void *));
	mergeq_free(mqp->mq_items, sizeof (mqp->mq_cap) * sizeof (void *));
	mqp->mq_items = items;
	mqp->mq_cap = ncap;
	return (0);
}

int
mergeq_add(mergeq_t *mqp, void *item)
{
	VERIFY0(mutex_lock(&mqp->mq_lock));
	if (mqp->mq_working == B_TRUE) {
		VERIFY0(mutex_unlock(&mqp->mq_lock));
		return (mergeq_error(ENXIO));
	}

	if (mqp->mq_next == mqp->mq_cap) {
		int ret;

		if ((ret = mergeq_grow(mqp)) != 0) {
			VERIFY0(mutex_unlock(&mqp->mq_lock));
			return (mergeq_error(ret));
		}
	}
	mqp->mq_items[mqp->mq_next] = item;
	mqp->mq_next++;
	mqp->mq_nitems++;

	VERIFY0(mutex_unlock(&mqp->mq_lock));
	return (0);
}

static size_t
mergeq_slot(mergeq_t *mqp)
{
	size_t s;

	VERIFY(MUTEX_HELD(&mqp->mq_lock));
	VERIFY(mqp->mq_next < mqp->mq_cap);

	/*
	 * This probably should be a cv / wait thing.
	 */
	VERIFY(mqp->mq_nproc != (mqp->mq_next + 1) % mqp->mq_cap);

	s = mqp->mq_next;
	mqp->mq_next++;
	if (mqp->mq_next == mqp->mq_cap) {
		mqp->mq_next %= mqp->mq_cap;
		mqp->mq_gnext++;
	}

	return (s);
}

/*
 * Internal function to push items onto the queue which is now a circular
 * buffer. This should only be used once we begin working on the queue.
 */
static void
mergeq_push(mergeq_t *mqp, size_t slot, void *item)
{
	VERIFY(MUTEX_HELD(&mqp->mq_lock));
	VERIFY(slot < mqp->mq_cap);

	/*
	 * We need to verify that we don't push over something that exists.
	 * Based on the design, this should never happen. However, in the face
	 * of bugs, anything is possible.
	 */
	while (mqp->mq_ncommit != slot && mqp->mq_iserror == B_FALSE)
		(void) cond_wait(&mqp->mq_cond, &mqp->mq_lock);

	if (mqp->mq_iserror == B_TRUE)
		return;

	mqp->mq_items[slot] = item;
	mqp->mq_nitems++;
	mqp->mq_ncommit++;
	if (mqp->mq_ncommit == mqp->mq_cap) {
		mqp->mq_ncommit %= mqp->mq_cap;
		mqp->mq_gncommit++;
	}
	(void) cond_broadcast(&mqp->mq_cond);
}

static void *
mergeq_pop_one(mergeq_t *mqp)
{
	void *out;

	/*
	 * We can't move mq_nproc beyond mq_next if they're on the same
	 * generation.
	 */
	VERIFY(mqp->mq_gnext != mqp->mq_gnproc ||
	    mqp->mq_nproc != mqp->mq_next);

	out = mqp->mq_items[mqp->mq_nproc];

	mqp->mq_items[mqp->mq_nproc] = NULL;
	mqp->mq_nproc++;
	if (mqp->mq_nproc == mqp->mq_cap) {
		mqp->mq_nproc %= mqp->mq_cap;
		mqp->mq_gnproc++;
	}
	mqp->mq_nitems--;

	return (out);
}

/*
 * Pop a set of two entries from the queue. We may not have anything to process
 * at the moment, eg. be waiting for someone to add something. In which case,
 * we'll be sitting and waiting.
 */
static boolean_t
mergeq_pop(mergeq_t *mqp, void **first, void **second)
{
	VERIFY(MUTEX_HELD(&mqp->mq_lock));
	VERIFY(mqp->mq_nproc < mqp->mq_cap);

	while (mqp->mq_nitems < 2 && mqp->mq_nactthrs > 0 &&
	    mqp->mq_iserror == B_FALSE)
		(void) cond_wait(&mqp->mq_cond, &mqp->mq_lock);

	if (mqp->mq_iserror == B_TRUE)
		return (B_FALSE);

	if (mqp->mq_nitems < 2 && mqp->mq_nactthrs == 0) {
		VERIFY(mqp->mq_iserror == B_TRUE || mqp->mq_nitems == 1);
		return (B_FALSE);
	}
	VERIFY(mqp->mq_nitems >= 2);

	*first = mergeq_pop_one(mqp);
	*second = mergeq_pop_one(mqp);

	return (B_TRUE);
}

static void *
mergeq_thr_merge(void *arg)
{
	mergeq_t *mqp = arg;

	VERIFY0(mutex_lock(&mqp->mq_lock));

	/*
	 * Check to make sure creation worked and if not, fail fast.
	 */
	if (mqp->mq_iserror == B_TRUE) {
		VERIFY0(mutex_unlock(&mqp->mq_lock));
		return (NULL);
	}

	for (;;) {
		void *first, *second, *out;
		int ret;
		size_t slot;

		if (mqp->mq_nitems == 1 && mqp->mq_nactthrs == 0) {
			VERIFY0(mutex_unlock(&mqp->mq_lock));
			return (NULL);
		}

		if (mergeq_pop(mqp, &first, &second) == B_FALSE) {
			VERIFY0(mutex_unlock(&mqp->mq_lock));
			return (NULL);
		}
		slot = mergeq_slot(mqp);

		mqp->mq_nactthrs++;

		VERIFY0(mutex_unlock(&mqp->mq_lock));
		ret = mqp->mq_func(first, second, &out, mqp->mq_arg);
		VERIFY0(mutex_lock(&mqp->mq_lock));

		if (ret != 0) {
			if (mqp->mq_iserror == B_FALSE) {
				mqp->mq_iserror = B_TRUE;
				mqp->mq_error = ret;
				(void) cond_broadcast(&mqp->mq_cond);
			}
			mqp->mq_nactthrs--;
			VERIFY0(mutex_unlock(&mqp->mq_lock));
			return (NULL);
		}
		mergeq_push(mqp, slot, out);
		mqp->mq_nactthrs--;
	}
}

int
mergeq_merge(mergeq_t *mqp, mergeq_proc_f *func, void *arg, void **outp,
    int *errp)
{
	int ret, i;
	boolean_t seterr = B_FALSE;

	if (mqp == NULL || func == NULL || outp == NULL) {
		return (mergeq_error(EINVAL));
	}

	VERIFY0(mutex_lock(&mqp->mq_lock));
	if (mqp->mq_working == B_TRUE) {
		VERIFY0(mutex_unlock(&mqp->mq_lock));
		return (mergeq_error(EBUSY));
	}

	if (mqp->mq_nitems == 0) {
		*outp = NULL;
		mergeq_reset(mqp);
		VERIFY0(mutex_unlock(&mqp->mq_lock));
		return (0);
	}

	/*
	 * Now that we've finished adding items to the queue, turn it into a
	 * circular buffer.
	 */
	mqp->mq_func = func;
	mqp->mq_arg = arg;
	mqp->mq_nproc = 0;
	mqp->mq_working = B_TRUE;
	if (mqp->mq_next == mqp->mq_cap) {
		mqp->mq_next %= mqp->mq_cap;
		mqp->mq_gnext++;
	}
	mqp->mq_ncommit = mqp->mq_next;

	ret = 0;
	for (i = 0; i < mqp->mq_ndthreads; i++) {
		ret = thr_create(NULL, 0, mergeq_thr_merge, mqp, 0,
		    &mqp->mq_thrs[i]);
		if (ret != 0) {
			mqp->mq_iserror = B_TRUE;
			break;
		}
	}

	VERIFY0(mutex_unlock(&mqp->mq_lock));
	if (ret == 0)
		(void) mergeq_thr_merge(mqp);

	for (i = 0; i < mqp->mq_ndthreads; i++) {
		VERIFY0(thr_join(mqp->mq_thrs[i], NULL, NULL));
	}

	VERIFY0(mutex_lock(&mqp->mq_lock));

	VERIFY(mqp->mq_nactthrs == 0);
	mqp->mq_working = B_FALSE;
	if (ret == 0 && mqp->mq_iserror == B_FALSE) {
		VERIFY(mqp->mq_nitems == 1);
		*outp = mergeq_pop_one(mqp);
	} else if (ret == 0 && mqp->mq_iserror == B_TRUE) {
		ret = MERGEQ_UERROR;
		if (errp != NULL)
			*errp = mqp->mq_error;
	} else {
		seterr = B_TRUE;
	}

	mergeq_reset(mqp);
	VERIFY0(mutex_unlock(&mqp->mq_lock));

	if (seterr == B_TRUE)
		return (mergeq_error(ret));

	return (ret);
}