/* * 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); }
