gdk_ssort.mx

这个是内存数据库中的一个管理工具

@' The contents of this file are subject to the MonetDB Public License
@' Version 1.1 (the "License"); you may not use this file except in
@' compliance with the License. You may obtain a copy of the License at
@' http://monetdb.cwi.nl/Legal/MonetDBLicense-1.1.html
@'
@' Software distributed under the License is distributed on an "AS IS"
@' basis, WITHOUT WARRANTY OF ANY KIND, either express or implied. See the
@' License for the specific language governing rights and limitations
@' under the License.
@'
@' The Original Code is the MonetDB Database System.
@'
@' The Initial Developer of the Original Code is CWI.
@' Portions created by CWI are Copyright (C) 1997-2007 CWI.
@' All Rights Reserved.

@f gdk_ssort
@a Sjoerd Mullender
@* Ssort
This file implements a stable sort algorithm.  The algorithm is a
straight copy of the listsort function in the Python 2.5 source code,
heavily modified to fit into the MonetDB environment.
@{
@c
#include "monetdb_config.h"
#include "gdk.h"

/* The maximum number of entries in a MergeState's pending-runs stack.
   This is enough to sort arrays of size up to about
       32 * phi ** MAX_MERGE_PENDING
   where phi ~= 1.618.  85 is ridiculously large enough, good for an array
   with 2**64 elements. */
#define MAX_MERGE_PENDING 85

/* When we get into galloping mode, we stay there until both runs win less
   often than MIN_GALLOP consecutive times.  See listsort.txt for more info. */
#define MIN_GALLOP 7

/* Avoid malloc for small temp arrays. */
#define MERGESTATE_TEMP_SIZE (256 * sizeof(void *))

/* One MergeState exists on the stack per invocation of mergesort.  It's just
   a convenient way to pass state around among the helper functions. */
struct slice {
	void *base;
	ssize_t len;
};

typedef struct {
	/* The comparison function. */
	int (*compare)(ptr, ptr);
	char *base;
	int width;
	int loc;
	/* Temporary storage for a single entry. If an entry is at
	   most 2 lng's, we don't need to allocate anything. */
	void *t;
	char tempstorage[2 * sizeof(lng)];

	/* This controls when we get *into* galloping mode.  It's
	   initialized to MIN_GALLOP.  merge_lo and merge_hi tend to
	   nudge it higher for random data, and lower for highly
	   structured data. */
	ssize_t min_gallop;

	/* 'a' is temp storage to help with merges.  It contains room
	   for alloced entries. */
	void **a;
	ssize_t alloced;

	/* A stack of n pending runs yet to be merged.  Run #i starts
	   at address base[i] and extends for len[i] elements.  It's
	   always true (so long as the indices are in bounds) that

		pending[i].base + pending[i].len == pending[i+1].base

	   so we could cut the storage for this, but it's a minor
	   amount, and keeping all the info explicit simplifies the
	   code. */
	int n;
	struct slice pending[MAX_MERGE_PENDING];

	/* 'a' points to this when possible, rather than muck with
	   malloc. */
	char temparray[MERGESTATE_TEMP_SIZE];
} MergeState;

/* Free all the temp memory owned by the MergeState.  This must be called
   when you're done with a MergeState, and may be called before then if
   you want to free the temp memory early. */
static void
merge_freemem(MergeState *ms)
{
	assert(ms != NULL);
	if (ms->a != (void *) ms->temparray)
		GDKfree(ms->a);
	ms->a = (void *) ms->temparray;
	ms->alloced = MERGESTATE_TEMP_SIZE;
}

/* Ensure enough temp memory for 'need' array slots is available.
   Returns 0 on success and -1 if the memory can't be gotten. */
static int
merge_getmem(MergeState *ms, ssize_t need)
{
	assert(ms != NULL);
	need *= ms->width;
	if (need <= ms->alloced)
		return 0;
	/* Don't realloc!  That can cost cycles to copy the old data, but
	   we don't care what's in the block.
	 */
	merge_freemem(ms);
	ms->a = GDKmalloc(need);
	if (ms->a) {
		ms->alloced = need;
		return 0;
	}
	GDKerror("GDKssort: not enough memory\n");
	merge_freemem(ms);	/* reset to sane state */
	return -1;
}

#define PTRADD(p, n, w)		(void *) ((char *) (p) + (n) * (w))

#define COPY(d,s,w) do {					\
		switch (w) {					\
		case sizeof(bte):				\
			* (bte *) d = * (bte *) s;		\
			break;					\
		case sizeof(sht):				\
			* (sht *) d = * (sht *) s;		\
			break;					\
		case sizeof(int):				\
			* (int *) d = * (int *) s;		\
			break;					\
		case sizeof(lng):				\
			* (lng *) d = * (lng *) s;		\
			break;					\
		case 2 * sizeof(lng):				\
			* (lng *) d = * (lng *) s;		\
			* ((lng *) d + 1) = * ((lng *) s + 1);	\
			break;					\
		default:					\
			memcpy(d, s, (size_t) w);		\
			break;					\
		}						\
	} while (0)

#define ISLT_any(X, Y, ms)  (((ms)->base ? (*(ms)->compare)((ms)->base + * (var_t *) PTRADD((X), (ms)->loc, 1), (ms)->base + * (var_t *) PTRADD((Y), (ms)->loc, 1)) : (*(ms)->compare)(PTRADD((X), (ms)->loc, 1), PTRADD((Y), (ms)->loc, 1))) < 0)
#define ISLT_any_rev(X, Y, ms)  (((ms)->base ? (*(ms)->compare)((ms)->base + * (var_t *) PTRADD((X), (ms)->loc, 1), (ms)->base + * (var_t *) PTRADD((Y), (ms)->loc, 1)) : (*(ms)->compare)(PTRADD((X), (ms)->loc, 1), PTRADD((Y), (ms)->loc, 1))) > 0)
@= islt
#define ISLT_@1@2(X, Y, ms)	(* (@1 *) PTRADD((X), (ms)->loc, 1) @3 * (@1 *) PTRADD((Y), (ms)->loc, 1))
@c
@:islt(chr,,<)@
@:islt(chr,_rev,>)@
@:islt(bte,,<)@
@:islt(bte,_rev,>)@
@:islt(sht,,<)@
@:islt(sht,_rev,>)@
@:islt(int,,<)@
@:islt(int,_rev,>)@
@:islt(lng,,<)@
@:islt(lng,_rev,>)@
@:islt(flt,,<)@
@:islt(flt,_rev,>)@
@:islt(dbl,,<)@
@:islt(dbl,_rev,>)@
@:islt(oid,,<)@
@:islt(oid,_rev,>)@

/* Reverse a slice of a list in place, from lo up to (exclusive) hi. */
static void
reverse_slice(void *lo, void *hi, MergeState *ms)
{
	void *t;
	int width;

	assert(lo && hi);
	assert(ms);

	t = ms->t;
	width = ms->width;

	hi = PTRADD(hi, -1, width);
	while (lo < hi) {
		COPY(t, lo, width);
		COPY(lo, hi, width);
		COPY(hi, t, width);
		lo = PTRADD(lo, 1, width);
		hi = PTRADD(hi, -1, width);
	}
}

#define MERGE_GETMEM(MS, NEED) ((NEED) * (MS)->width <= (MS)->alloced ? 0 :	\
				merge_getmem(MS, NEED))

@= ssort
/* binarysort is the best method for sorting small arrays: it does
   few compares, but can do data movement quadratic in the number of
   elements.
   [lo, hi) is a contiguous slice of a list, and is sorted via
   binary insertion.  This sort is stable.
   On entry, must have lo <= start <= hi, and that [lo, start) is already
   sorted (pass start == lo if you don't know!). */
static void
binarysort_@1@2(void *lo, void *hi, void *start, MergeState *ms)
{
	register void *l, *p, *r;

	assert(lo <= start && start <= hi);
	/* assert [lo, start) is sorted */
	if (lo == start)
		start = PTRADD(start, 1, ms->width);
	/* [lo,start) is sorted, insert start (the pivot) into this
	   area, finding its position using binary search. */
	for (; start < hi; start = PTRADD(start, 1, ms->width)) {
		/* set l to where start belongs */
		l = lo;
		r = start;
		/* ms->t is the pivot */
		COPY(ms->t, r, ms->width);
		/* Invariants:
		   pivot >= all in [lo, l).
		   pivot  < all in [r, start).
		   The second is vacuously true at the start.
		 */
		assert(l < r);
		do {
			p = PTRADD(l, (((char *) r - (char *) l) / ms->width) >> 1, ms->width);
			if (ISLT_@1@2(ms->t, p, ms))
				r = p;
			else
				l = PTRADD(p, 1, ms->width);
		} while (l < r);
		assert(l == r);
		/* The invariants still hold, so pivot >= all in [lo, l) and
		   pivot < all in [l, start), so pivot belongs at l.  Note
		   that if there are elements equal to pivot, l points to the
		   first slot after them -- that's why this sort is stable.
		   Slide over to make room.
		   Caution: using memmove is much slower under MSVC 5;
		   we're not usually moving many slots. */
		for (p = start, r = PTRADD(p, -1, ms->width); p > l; p = r, r = PTRADD(p, -1, ms->width))
			COPY(p, r, ms->width);
		COPY(l, ms->t, ms->width);
	}
}

/* Locate the proper position of key in a sorted vector; if the vector contains
   an element equal to key, return the position immediately to the left of
   the leftmost equal element.  [gallop_right() does the same except returns
   the position to the right of the rightmost equal element (if any).]

   "a" is a sorted vector with n elements, starting at a[0].  n must be > 0.

   "hint" is an index at which to begin the search, 0 <= hint < n.  The closer
   hint is to the final result, the faster this runs.

   The return value is the int k in 0..n such that

	a[k-1] < key <= a[k]

   pretending that *(a-1) is minus infinity and a[n] is plus infinity.  IOW,
   key belongs at index k; or, IOW, the first k elements of a should precede
   key, and the last n-k should follow key.

   Returns -1 on error.  See listsort.txt for info on the method. */
static ssize_t
gallop_left_@1@2(void *key, void *a, ssize_t n, ssize_t hint, MergeState *ms)
{
	ssize_t ofs;
	ssize_t lastofs;
	ssize_t k;

	assert(key && a && n > 0 && hint >= 0 && hint < n);

	a = PTRADD(a, hint, ms->width);
	lastofs = 0;
	ofs = 1;
	if (ISLT_@1@2(a, key, ms)) {
		/* a[hint] < key -- gallop right, until
		   a[hint + lastofs] < key <= a[hint + ofs]
		 */
		const ssize_t maxofs = n - hint;	/* &a[n-1] is highest */
		while (ofs < maxofs) {
			if (ISLT_@1@2(PTRADD(a, ofs, ms->width), key, ms)) {
				lastofs = ofs;
				ofs = (ofs << 1) + 1;
				if (ofs <= 0)	/* int overflow */
					ofs = maxofs;
			} else	/* key <= a[hint + ofs] */
				break;
		}
		if (ofs > maxofs)
			ofs = maxofs;
		/* Translate back to offsets relative to &a[0]. */
		lastofs += hint;
		ofs += hint;
	} else {
		/* key <= a[hint] -- gallop left, until
		   a[hint - ofs] < key <= a[hint - lastofs]
		 */
		const ssize_t maxofs = hint + 1;	/* &a[0] is lowest */
		while (ofs < maxofs) {
			if (ISLT_@1@2(PTRADD(a, -ofs, ms->width), key, ms))
				break;
			/* key <= a[hint - ofs] */
			lastofs = ofs;
			ofs = (ofs << 1) + 1;
			if (ofs <= 0)	/* int overflow */
				ofs = maxofs;
		}
		if (ofs > maxofs)
			ofs = maxofs;
		/* Translate back to positive offsets relative to &a[0]. */
		k = lastofs;
		lastofs = hint - ofs;
		ofs = hint - k;
	}
	a = PTRADD(a, -hint, ms->width);

	assert(-1 <= lastofs && lastofs < ofs && ofs <= n);
	/* Now a[lastofs] < key <= a[ofs], so key belongs somewhere to the
	   right of lastofs but no farther right than ofs.  Do a binary
	   search, with invariant a[lastofs-1] < key <= a[ofs].
	 */
	++lastofs;
	while (lastofs < ofs) {
		ssize_t m = lastofs + ((ofs - lastofs) >> 1);

		if (ISLT_@1@2(PTRADD(a, m, ms->width), key, ms))
			lastofs = m + 1; /* a[m] < key */
		else
			ofs = m;	/* key <= a[m] */
	}
	assert(lastofs == ofs);		/* so a[ofs-1] < key <= a[ofs] */
	return ofs;
}

/* Exactly like gallop_left(), except that if key already exists in a[0:n],
   finds the position immediately to the right of the rightmost equal value.

   The return value is the int k in 0..n such that

	a[k-1] <= key < a[k]

   or -1 if error.

   The code duplication is massive, but this is enough different given that
   we're sticking to "<" comparisons that it's much harder to follow if
   written as one routine with yet another "left or right?" flag. */
static ssize_t
gallop_right_@1@2(void *key, void *a, ssize_t n, ssize_t hint, MergeState *ms)
{
	ssize_t ofs;
	ssize_t lastofs;
	ssize_t k;

	assert(key && a && n > 0 && hint >= 0 && hint < n);

	a = PTRADD(a, hint, ms->width);
	lastofs = 0;
	ofs = 1;
	if (ISLT_@1@2(key, a, ms)) {
		/* key < a[hint] -- gallop left, until
		   a[hint - ofs] <= key < a[hint - lastofs]
		 */
		const ssize_t maxofs = hint + 1;	/* &a[0] is lowest */
		while (ofs < maxofs) {
			if (ISLT_@1@2(key, PTRADD(a, -ofs, ms->width), ms)) {
				lastofs = ofs;
				ofs = (ofs << 1) + 1;
				if (ofs <= 0)	/* int overflow */
					ofs = maxofs;
			} else	/* a[hint - ofs] <= key */
				break;
		}
		if (ofs > maxofs)
			ofs = maxofs;
		/* Translate back to positive offsets relative to &a[0]. */
		k = lastofs;
		lastofs = hint - ofs;
		ofs = hint - k;
	} else {
		/* a[hint] <= key -- gallop right, until
		   a[hint + lastofs] <= key < a[hint + ofs]
		*/
		const ssize_t maxofs = n - hint;	/* &a[n-1] is highest */
		while (ofs < maxofs) {
			if (ISLT_@1@2(key, PTRADD(a, ofs, ms->width), ms))
				break;
			/* a[hint + ofs] <= key */
			lastofs = ofs;
			ofs = (ofs << 1) + 1;
			if (ofs <= 0)	/* int overflow */
				ofs = maxofs;
		}
		if (ofs > maxofs)
			ofs = maxofs;
		/* Translate back to offsets relative to &a[0]. */
		lastofs += hint;
		ofs += hint;
	}
	a = PTRADD(a, -hint, ms->width);

	assert(-1 <= lastofs && lastofs < ofs && ofs <= n);
	/* Now a[lastofs] <= key < a[ofs], so key belongs somewhere to the
	   right of lastofs but no farther right than ofs.  Do a binary
	   search, with invariant a[lastofs-1] <= key < a[ofs].
	 */
	++lastofs;
	while (lastofs < ofs) {
		ssize_t m = lastofs + ((ofs - lastofs) >> 1);

		if (ISLT_@1@2(key, PTRADD(a, m, ms->width), ms))
			ofs = m;	/* key < a[m] */
		else
			lastofs = m+1;	/* a[m] <= key */
	}
	assert(lastofs == ofs);		/* so a[ofs-1] <= key < a[ofs] */
	return ofs;
}

/* Merge the two runs at stack indices i and i+1.
   Returns 0 on success, -1 on error. */
static ssize_t
merge_at_@1@2(MergeState *ms, ssize_t i)
{
	void *pa, *pb;
	ssize_t na, nb;
	ssize_t k;

	assert(ms != NULL);
	assert(ms->n >= 2);
	assert(i >= 0);
	assert(i == ms->n - 2 || i == ms->n - 3);

	pa = ms->pending[i].base;
	na = ms->pending[i].len;
	pb = ms->pending[i + 1].base;
	nb = ms->pending[i + 1].len;
	assert(na > 0 && nb > 0);
	assert(PTRADD(pa, na, ms->width) == pb);

	/* Record the length of the combined runs; if i is the 3rd-last
	   run now, also slide over the last run (which isn't involved
	   in this merge).  The current run i+1 goes away in any case.
	 */
	ms->pending[i].len = na + nb;
	if (i == ms->n - 3)
		ms->pending[i + 1] = ms->pending[i + 2];
	--ms->n;

	/* Where does b start in a?  Elements in a before that can be
	   ignored (already in place).
	 */
	k = gallop_right_@1@2(pb, pa, na, 0, ms);
	pa = PTRADD(pa, k, ms->width);
	na -= k;
	if (na == 0)
		return 0;

	/* Where does a end in b?  Elements in b after that can be
	   ignored (already in place).
	 */
	nb = gallop_left_@1@2(PTRADD(pa, na - 1, ms->width), pb, nb, nb-1, ms);
	if (nb <= 0)
		return nb;

	/* Merge what remains of the runs, using a temp array with
	   min(na, nb) elements.
	 */
	if (na <= nb) {
/* Merge the na elements starting at pa with the nb elements starting at pb
   in a stable way, in-place.  na and nb must be > 0, and pa + na == pb.
   Must also have that *pb < *pa, that pa[na-1] belongs at the end of the
   merge, and should have na <= nb.  See listsort.txt for more info.
   Return 0 if successful, -1 if error. */
		void *dest;
		ssize_t min_gallop = ms->min_gallop;

		assert(ms && pa && pb && na > 0 && nb > 0 && PTRADD(pa, na, ms->width) == pb);
		if (MERGE_GETMEM(ms, na) < 0)
			return -1;
		COPY(ms->a, pa, na * ms->width);
		dest = pa;
		pa = ms->a;

		COPY(dest, pb, ms->width);
		dest = PTRADD(dest, 1, ms->width);
		pb = PTRADD(pb, 1, ms->width);
		--nb;
		if (nb == 0)
			goto SucceedA;
		if (na == 1)
			goto CopyB;

		for (;;) {
			ssize_t acount = 0;	/* # of times A won in a row */
			ssize_t bcount = 0;	/* # of times B won in a row */

			/* Do the straightforward thing until (if ever) one run
			   appears to win consistently.