This is a live mirror of the Perl 5 development currently hosted at https://github.com/perl/perl5
fix IO::File to support binmode
[perl5.git] / pp_sort.c
index 844c0e3..0cb4795 100644 (file)
--- a/pp_sort.c
+++ b/pp_sort.c
@@ -1,6 +1,7 @@
 /*    pp_sort.c
  *
- *    Copyright (c) 1991-2001, Larry Wall
+ *    Copyright (C) 1991, 1992, 1993, 1994, 1995, 1996, 1997, 1998, 1999,
+ *    2000, 2001, 2002, 2003, 2004, by Larry Wall and others
  *
  *    You may distribute under the terms of either the GNU General Public
  *    License or the Artistic License, as specified in the README file.
  *   rear!'  the slave-driver shouted. 'Three files up. And stay there...
  */
 
+/* This file contains pp ("push/pop") functions that
+ * execute the opcodes that make up a perl program. A typical pp function
+ * expects to find its arguments on the stack, and usually pushes its
+ * results onto the stack, hence the 'pp' terminology. Each OP structure
+ * contains a pointer to the relevant pp_foo() function.
+ *
+ * This particular file just contains pp_sort(), which is complex
+ * enough to merit its own file! See the other pp*.c files for the rest of
+ * the pp_ functions.
+ */
+
 #include "EXTERN.h"
 #define PERL_IN_PP_SORT_C
 #include "perl.h"
 
+#if defined(UNDER_CE)
+/* looks like 'small' is reserved word for WINCE (or somesuch)*/
+#define        small xsmall
+#endif
+
 static I32 sortcv(pTHX_ SV *a, SV *b);
 static I32 sortcv_stacked(pTHX_ SV *a, SV *b);
 static I32 sortcv_xsub(pTHX_ SV *a, SV *b);
@@ -29,10 +46,13 @@ static I32 amagic_cmp_locale(pTHX_ SV *a, SV *b);
 #define sv_cmp_static Perl_sv_cmp
 #define sv_cmp_locale_static Perl_sv_cmp_locale
 
-#define SORTHINTS(hintsvp) \
-     ((PL_hintgv &&    \
-      (hintsvp = hv_fetch(GvHV(PL_hintgv), "SORT", 4, FALSE))) ? \
-         (I32)SvIV(*hintsvp) : 0)
+#define SORTHINTS(hintsv) \
+    (((hintsv) = GvSV(gv_fetchpv("sort::hints", GV_ADDMULTI, SVt_IV))), \
+    (SvIOK(hintsv) ? ((I32)SvIV(hintsv)) : 0))
+
+#ifndef SMALLSORT
+#define        SMALLSORT (200)
+#endif
 
 /*
  * The mergesort implementation is by Peter M. Mcilroy <pmcilroy@lucent.com>.
@@ -48,15 +68,6 @@ static I32 amagic_cmp_locale(pTHX_ SV *a, SV *b);
  *
  */
 
-#ifdef TESTHARNESS
-#include <sys/types.h>
-typedef        void SV;
-#define pTHX_
-#define STATIC
-#define New(ID,VAR,N,TYPE) VAR=(TYPE *)malloc((N)*sizeof(TYPE))
-#define        Safefree(VAR) free(VAR)
-typedef int  (*SVCOMPARE_t) (pTHX_ SV*, SV*);
-#endif /* TESTHARNESS */
 
 typedef char * aptr;           /* pointer for arithmetic on sizes */
 typedef SV * gptr;             /* pointers in our lists */
@@ -177,13 +188,14 @@ typedef SV * gptr;                /* pointers in our lists */
 */
 
 
-static void
+static IV
 dynprep(pTHX_ gptr *list1, gptr *list2, size_t nmemb, SVCOMPARE_t cmp)
 {
-    int sense;
+    I32 sense;
     register gptr *b, *p, *q, *t, *p2;
     register gptr c, *last, *r;
     gptr *savep;
+    IV runs = 0;
 
     b = list1;
     last = PINDEX(b, nmemb);
@@ -221,7 +233,7 @@ dynprep(pTHX_ gptr *list1, gptr *list2, size_t nmemb, SVCOMPARE_t cmp)
                    ((t + 1) == last) &&
                    ((cmp(aTHX_ *(p-1), *p) > 0) == sense))
                    savep = r = p = q = last;
-               p2 = NEXT(p2) = p2 + (p - b);
+               p2 = NEXT(p2) = p2 + (p - b); ++runs;
                if (sense) while (b < --p) {
                    c = *b;
                    *b++ = *p;
@@ -230,7 +242,7 @@ dynprep(pTHX_ gptr *list1, gptr *list2, size_t nmemb, SVCOMPARE_t cmp)
                p = savep;
            }
            while (q < p) {             /* simple pairs */
-               p2 = NEXT(p2) = p2 + 2;
+               p2 = NEXT(p2) = p2 + 2; ++runs;
                if (sense) {
                    c = *q++;
                    *(q-1) = *q;
@@ -238,164 +250,316 @@ dynprep(pTHX_ gptr *list1, gptr *list2, size_t nmemb, SVCOMPARE_t cmp)
                } else q += 2;
            }
            if (((b = p) == t) && ((t+1) == last)) {
-               NEXT(p2) = p2 + 1;
+               NEXT(p2) = p2 + 1; ++runs;
                b++;
            }
            q = r;
        } while (b < t);
        sense = !sense;
     }
-    return;
+    return runs;
 }
 
 
-/* Overview of bmerge variables:
-**
-** list1 and list2 address the main and auxiliary arrays.
-** They swap identities after each merge pass.
-** Base points to the original list1, so we can tell if
-** the pointers ended up where they belonged (or must be copied).
-**
-** When we are merging two lists, f1 and f2 are the next elements
-** on the respective lists.  l1 and l2 mark the end of the lists.
-** tp2 is the current location in the merged list.
-**
-** p1 records where f1 started.
-** After the merge, a new descriptor is built there.
-**
-** p2 is a ``parallel'' pointer in (what starts as) descriptor space.
-** It is used to identify and delimit the runs.
-**
-** In the heat of determining where q, the greater of the f1/f2 elements,
-** belongs in the other list, b, t and p, represent bottom, top and probe
-** locations, respectively, in the other list.
-** They make convenient temporary pointers in other places.
-*/
+/* The original merge sort, in use since 5.7, was as fast as, or faster than,
+ * qsort on many platforms, but slower than qsort, conspicuously so,
+ * on others.  The most likely explanation was platform-specific
+ * differences in cache sizes and relative speeds.
+ *
+ * The quicksort divide-and-conquer algorithm guarantees that, as the
+ * problem is subdivided into smaller and smaller parts, the parts
+ * fit into smaller (and faster) caches.  So it doesn't matter how
+ * many levels of cache exist, quicksort will "find" them, and,
+ * as long as smaller is faster, take advanatge of them.
+ *
+ * By contrast, consider how the original mergesort algorithm worked.
+ * Suppose we have five runs (each typically of length 2 after dynprep).
+ * 
+ * pass               base                        aux
+ *  0              1 2 3 4 5
+ *  1                                           12 34 5
+ *  2                1234 5
+ *  3                                            12345
+ *  4                 12345
+ *
+ * Adjacent pairs are merged in "grand sweeps" through the input.
+ * This means, on pass 1, the records in runs 1 and 2 aren't revisited until
+ * runs 3 and 4 are merged and the runs from run 5 have been copied.
+ * The only cache that matters is one large enough to hold *all* the input.
+ * On some platforms, this may be many times slower than smaller caches.
+ *
+ * The following pseudo-code uses the same basic merge algorithm,
+ * but in a divide-and-conquer way.
+ *
+ * # merge $runs runs at offset $offset of list $list1 into $list2.
+ * # all unmerged runs ($runs == 1) originate in list $base.
+ * sub mgsort2 {
+ *     my ($offset, $runs, $base, $list1, $list2) = @_;
+ *
+ *     if ($runs == 1) {
+ *         if ($list1 is $base) copy run to $list2
+ *         return offset of end of list (or copy)
+ *     } else {
+ *         $off2 = mgsort2($offset, $runs-($runs/2), $base, $list2, $list1)
+ *         mgsort2($off2, $runs/2, $base, $list2, $list1)
+ *         merge the adjacent runs at $offset of $list1 into $list2
+ *         return the offset of the end of the merged runs
+ *     }
+ * }
+ * mgsort2(0, $runs, $base, $aux, $base);
+ *
+ * For our 5 runs, the tree of calls looks like 
+ *
+ *           5
+ *      3        2
+ *   2     1   1   1
+ * 1   1
+ *
+ * 1   2   3   4   5
+ *
+ * and the corresponding activity looks like
+ *
+ * copy runs 1 and 2 from base to aux
+ * merge runs 1 and 2 from aux to base
+ * (run 3 is where it belongs, no copy needed)
+ * merge runs 12 and 3 from base to aux
+ * (runs 4 and 5 are where they belong, no copy needed)
+ * merge runs 4 and 5 from base to aux
+ * merge runs 123 and 45 from aux to base
+ *
+ * Note that we merge runs 1 and 2 immediately after copying them,
+ * while they are still likely to be in fast cache.  Similarly,
+ * run 3 is merged with run 12 while it still may be lingering in cache.
+ * This implementation should therefore enjoy much of the cache-friendly
+ * behavior that quicksort does.  In addition, it does less copying
+ * than the original mergesort implementation (only runs 1 and 2 are copied)
+ * and the "balancing" of merges is better (merged runs comprise more nearly
+ * equal numbers of original runs).
+ *
+ * The actual cache-friendly implementation will use a pseudo-stack
+ * to avoid recursion, and will unroll processing of runs of length 2,
+ * but it is otherwise similar to the recursive implementation.
+ */
+
+typedef struct {
+    IV offset;         /* offset of 1st of 2 runs at this level */
+    IV runs;           /* how many runs must be combined into 1 */
+} off_runs;            /* pseudo-stack element */
+
+
+static I32
+cmp_desc(pTHX_ gptr a, gptr b)
+{
+    return -PL_sort_RealCmp(aTHX_ a, b);
+}
 
 STATIC void
-S_mergesortsv(pTHX_ gptr *list1, size_t nmemb, SVCOMPARE_t cmp)
+S_mergesortsv(pTHX_ gptr *base, size_t nmemb, SVCOMPARE_t cmp, U32 flags)
 {
-    int i, run;
-    int sense;
+    IV i, run, runs, offset;
+    I32 sense, level;
+    int iwhich;
     register gptr *f1, *f2, *t, *b, *p, *tp2, *l1, *l2, *q;
-    gptr *aux, *list2, *p2, *last;
-    gptr *base = list1;
+    gptr *aux, *list1, *list2;
     gptr *p1;
+    gptr small[SMALLSORT];
+    gptr *which[3];
+    off_runs stack[60], *stackp;
+    SVCOMPARE_t savecmp;
 
-    if (nmemb <= 1) return;    /* sorted trivially */
-    New(799,list2,nmemb,gptr); /* allocate auxilliary array */
-    aux = list2;
-    dynprep(aTHX_ list1, list2, nmemb, cmp);
-    last = PINDEX(list2, nmemb);
-    while (NEXT(list2) != last) {
-       /* More than one run remains.  Do some merging to reduce runs. */
-       l2 = p1 = list1;
-       for (tp2 = p2 = list2; p2 != last;) {
-           /* The new first run begins where the old second list ended.
-           ** Use the p2 ``parallel'' pointer to identify the end of the run.
-           */
-           f1 = l2;
-           t = NEXT(p2);
-           f2 = l1 = POTHER(t, list2, list1);
-           if (t != last) t = NEXT(t);
-           l2 = POTHER(t, list2, list1);
-           p2 = t;
-           while (f1 < l1 && f2 < l2) {
-               /* If head 1 is larger than head 2, find ALL the elements
-               ** in list 2 strictly less than head1, write them all,
-               ** then head 1.  Then compare the new heads, and repeat,
-               ** until one or both lists are exhausted.
-               **
-               ** In all comparisons (after establishing
-               ** which head to merge) the item to merge
-               ** (at pointer q) is the first operand of
-               ** the comparison.  When we want to know
-               ** if ``q is strictly less than the other'',
-               ** we can't just do
-               **    cmp(q, other) < 0
-               ** because stability demands that we treat equality
-               ** as high when q comes from l2, and as low when
-               ** q was from l1.  So we ask the question by doing
-               **    cmp(q, other) <= sense
-               ** and make sense == 0 when equality should look low,
-               ** and -1 when equality should look high.
-               */
-
-
-               if (cmp(aTHX_ *f1, *f2) <= 0) {
-                   q = f2; b = f1; t = l1;
-                   sense = -1;
-               } else {
-                   q = f1; b = f2; t = l2;
-                   sense = 0;
-               }
+    if (nmemb <= 1) return;                    /* sorted trivially */
 
+    if (flags) {
+       savecmp = PL_sort_RealCmp;      /* Save current comparison routine, if any */
+       PL_sort_RealCmp = cmp;  /* Put comparison routine where cmp_desc can find it */
+       cmp = cmp_desc;
+    }
 
-               /* ramp up
-               **
-               ** Leave t at something strictly
-               ** greater than q (or at the end of the list),
-               ** and b at something strictly less than q.
-               */
-               for (i = 1, run = 0 ;;) {
-                   if ((p = PINDEX(b, i)) >= t) {
-                       /* off the end */
-                       if (((p = PINDEX(t, -1)) > b) &&
-                           (cmp(aTHX_ *q, *p) <= sense))
-                            t = p;
-                       else b = p;
-                       break;
-                   } else if (cmp(aTHX_ *q, *p) <= sense) {
-                       t = p;
-                       break;
-                   } else b = p;
-                   if (++run >= RTHRESH) i += i;
-               }
+    if (nmemb <= SMALLSORT) aux = small;       /* use stack for aux array */
+    else { New(799,aux,nmemb,gptr); }          /* allocate auxilliary array */
+    level = 0;
+    stackp = stack;
+    stackp->runs = dynprep(aTHX_ base, aux, nmemb, cmp);
+    stackp->offset = offset = 0;
+    which[0] = which[2] = base;
+    which[1] = aux;
+    for (;;) {
+       /* On levels where both runs have be constructed (stackp->runs == 0),
+        * merge them, and note the offset of their end, in case the offset
+        * is needed at the next level up.  Hop up a level, and,
+        * as long as stackp->runs is 0, keep merging.
+        */
+       if ((runs = stackp->runs) == 0) {
+           iwhich = level & 1;
+           list1 = which[iwhich];              /* area where runs are now */
+           list2 = which[++iwhich];            /* area for merged runs */
+           do {
+               offset = stackp->offset;
+               f1 = p1 = list1 + offset;               /* start of first run */
+               p = tp2 = list2 + offset;       /* where merged run will go */
+               t = NEXT(p);                    /* where first run ends */
+               f2 = l1 = POTHER(t, list2, list1); /* ... on the other side */
+               t = NEXT(t);                    /* where second runs ends */
+               l2 = POTHER(t, list2, list1);   /* ... on the other side */
+               offset = PNELEM(list2, t);
+               while (f1 < l1 && f2 < l2) {
+                   /* If head 1 is larger than head 2, find ALL the elements
+                   ** in list 2 strictly less than head1, write them all,
+                   ** then head 1.  Then compare the new heads, and repeat,
+                   ** until one or both lists are exhausted.
+                   **
+                   ** In all comparisons (after establishing
+                   ** which head to merge) the item to merge
+                   ** (at pointer q) is the first operand of
+                   ** the comparison.  When we want to know
+                   ** if ``q is strictly less than the other'',
+                   ** we can't just do
+                   **    cmp(q, other) < 0
+                   ** because stability demands that we treat equality
+                   ** as high when q comes from l2, and as low when
+                   ** q was from l1.  So we ask the question by doing
+                   **    cmp(q, other) <= sense
+                   ** and make sense == 0 when equality should look low,
+                   ** and -1 when equality should look high.
+                   */
 
 
-               /* q is known to follow b and must be inserted before t.
-               ** Increment b, so the range of possibilities is [b,t).
-               ** Round binary split down, to favor early appearance.
-               ** Adjust b and t until q belongs just before t.
-               */
+                   if (cmp(aTHX_ *f1, *f2) <= 0) {
+                       q = f2; b = f1; t = l1;
+                       sense = -1;
+                   } else {
+                       q = f1; b = f2; t = l2;
+                       sense = 0;
+                   }
 
-               b++;
-               while (b < t) {
-                   p = PINDEX(b, (PNELEM(b, t) - 1) / 2);
-                   if (cmp(aTHX_ *q, *p) <= sense) {
-                       t = p;
-                   } else b = p + 1;
-               }
 
+                   /* ramp up
+                   **
+                   ** Leave t at something strictly
+                   ** greater than q (or at the end of the list),
+                   ** and b at something strictly less than q.
+                   */
+                   for (i = 1, run = 0 ;;) {
+                       if ((p = PINDEX(b, i)) >= t) {
+                           /* off the end */
+                           if (((p = PINDEX(t, -1)) > b) &&
+                               (cmp(aTHX_ *q, *p) <= sense))
+                                t = p;
+                           else b = p;
+                           break;
+                       } else if (cmp(aTHX_ *q, *p) <= sense) {
+                           t = p;
+                           break;
+                       } else b = p;
+                       if (++run >= RTHRESH) i += i;
+                   }
+
+
+                   /* q is known to follow b and must be inserted before t.
+                   ** Increment b, so the range of possibilities is [b,t).
+                   ** Round binary split down, to favor early appearance.
+                   ** Adjust b and t until q belongs just before t.
+                   */
+
+                   b++;
+                   while (b < t) {
+                       p = PINDEX(b, (PNELEM(b, t) - 1) / 2);
+                       if (cmp(aTHX_ *q, *p) <= sense) {
+                           t = p;
+                       } else b = p + 1;
+                   }
 
-               /* Copy all the strictly low elements */
 
-               if (q == f1) {
-                   FROMTOUPTO(f2, tp2, t);
-                   *tp2++ = *f1++;
-               } else {
-                   FROMTOUPTO(f1, tp2, t);
-                   *tp2++ = *f2++;
+                   /* Copy all the strictly low elements */
+
+                   if (q == f1) {
+                       FROMTOUPTO(f2, tp2, t);
+                       *tp2++ = *f1++;
+                   } else {
+                       FROMTOUPTO(f1, tp2, t);
+                       *tp2++ = *f2++;
+                   }
                }
-           }
 
 
-           /* Run out remaining list */
-           if (f1 == l1) {
-                  if (f2 < l2) FROMTOUPTO(f2, tp2, l2);
-           } else              FROMTOUPTO(f1, tp2, l1);
-           p1 = NEXT(p1) = POTHER(tp2, list2, list1);
+               /* Run out remaining list */
+               if (f1 == l1) {
+                      if (f2 < l2) FROMTOUPTO(f2, tp2, l2);
+               } else              FROMTOUPTO(f1, tp2, l1);
+               p1 = NEXT(p1) = POTHER(tp2, list2, list1);
+
+               if (--level == 0) goto done;
+               --stackp;
+               t = list1; list1 = list2; list2 = t;    /* swap lists */
+           } while ((runs = stackp->runs) == 0);
+       }
+
+
+       stackp->runs = 0;               /* current run will finish level */
+       /* While there are more than 2 runs remaining,
+        * turn them into exactly 2 runs (at the "other" level),
+        * each made up of approximately half the runs.
+        * Stack the second half for later processing,
+        * and set about producing the first half now.
+        */
+       while (runs > 2) {
+           ++level;
+           ++stackp;
+           stackp->offset = offset;
+           runs -= stackp->runs = runs / 2;
+       }
+       /* We must construct a single run from 1 or 2 runs.
+        * All the original runs are in which[0] == base.
+        * The run we construct must end up in which[level&1].
+        */
+       iwhich = level & 1;
+       if (runs == 1) {
+           /* Constructing a single run from a single run.
+            * If it's where it belongs already, there's nothing to do.
+            * Otherwise, copy it to where it belongs.
+            * A run of 1 is either a singleton at level 0,
+            * or the second half of a split 3.  In neither event
+            * is it necessary to set offset.  It will be set by the merge
+            * that immediately follows.
+            */
+           if (iwhich) {       /* Belongs in aux, currently in base */
+               f1 = b = PINDEX(base, offset);  /* where list starts */
+               f2 = PINDEX(aux, offset);       /* where list goes */
+               t = NEXT(f2);                   /* where list will end */
+               offset = PNELEM(aux, t);        /* offset thereof */
+               t = PINDEX(base, offset);       /* where it currently ends */
+               FROMTOUPTO(f1, f2, t);          /* copy */
+               NEXT(b) = t;                    /* set up parallel pointer */
+           } else if (level == 0) goto done;   /* single run at level 0 */
+       } else {
+           /* Constructing a single run from two runs.
+            * The merge code at the top will do that.
+            * We need only make sure the two runs are in the "other" array,
+            * so they'll end up in the correct array after the merge.
+            */
+           ++level;
+           ++stackp;
+           stackp->offset = offset;
+           stackp->runs = 0;   /* take care of both runs, trigger merge */
+           if (!iwhich) {      /* Merged runs belong in aux, copy 1st */
+               f1 = b = PINDEX(base, offset);  /* where first run starts */
+               f2 = PINDEX(aux, offset);       /* where it will be copied */
+               t = NEXT(f2);                   /* where first run will end */
+               offset = PNELEM(aux, t);        /* offset thereof */
+               p = PINDEX(base, offset);       /* end of first run */
+               t = NEXT(t);                    /* where second run will end */
+               t = PINDEX(base, PNELEM(aux, t)); /* where it now ends */
+               FROMTOUPTO(f1, f2, t);          /* copy both runs */
+               NEXT(b) = p;                    /* paralled pointer for 1st */
+               NEXT(p) = t;                    /* ... and for second */
+           }
        }
-       t = list1;
-       list1 = list2;
-       list2 = t;
-       last = PINDEX(list2, nmemb);
     }
-    if (base == list2) {
-       last = PINDEX(list1, nmemb);
-       FROMTOUPTO(list1, list2, last);
+done:
+    if (aux != small) Safefree(aux);   /* free iff allocated */
+    if (flags) {
+        PL_sort_RealCmp = savecmp;     /* Restore current comparison routine, if any */
     }
-    Safefree(aux);
     return;
 }
 
@@ -623,7 +787,7 @@ S_qsortsvu(pTHX_ SV ** array, size_t num_elts, SVCOMPARE_t compare)
       register size_t n, j;
       register SV **q;
       for (n = num_elts, q = array; n > 1; ) {
-         j = n-- * Drand01();
+         j = (size_t)(n-- * Drand01());
          temp = q[j];
          q[j] = q[n];
          q[n] = temp;
@@ -1101,10 +1265,6 @@ S_qsortsvu(pTHX_ SV ** array, size_t num_elts, SVCOMPARE_t compare)
    /* Believe it or not, the array is sorted at this point! */
 }
 
-#ifndef SMALLSORT
-#define        SMALLSORT (200)
-#endif
-
 /* Stabilize what is, presumably, an otherwise unstable sort method.
  * We do that by allocating (or having on hand) an array of pointers
  * that is the same size as the original array of elements to be sorted.
@@ -1156,7 +1316,6 @@ S_qsortsvu(pTHX_ SV ** array, size_t num_elts, SVCOMPARE_t compare)
  * dictated by the indirect array.
  */
 
-static SVCOMPARE_t RealCmp;
 
 static I32
 cmpindir(pTHX_ gptr a, gptr b)
@@ -1165,19 +1324,32 @@ cmpindir(pTHX_ gptr a, gptr b)
     gptr *ap = (gptr *)a;
     gptr *bp = (gptr *)b;
 
-    if ((sense = RealCmp(aTHX_ *ap, *bp)) == 0)
+    if ((sense = PL_sort_RealCmp(aTHX_ *ap, *bp)) == 0)
         sense = (ap > bp) ? 1 : ((ap < bp) ? -1 : 0);
     return sense;
 }
 
+static I32
+cmpindir_desc(pTHX_ gptr a, gptr b)
+{
+    I32 sense;
+    gptr *ap = (gptr *)a;
+    gptr *bp = (gptr *)b;
+
+    /* Reverse the default */
+    if ((sense = PL_sort_RealCmp(aTHX_ *ap, *bp)))
+       return -sense;
+    /* But don't reverse the stability test.  */
+    return (ap > bp) ? 1 : ((ap < bp) ? -1 : 0);
+
+}
+
 STATIC void
-S_qsortsv(pTHX_ gptr *list1, size_t nmemb, SVCOMPARE_t cmp)
+S_qsortsv(pTHX_ gptr *list1, size_t nmemb, SVCOMPARE_t cmp, U32 flags)
 {
-    SV **hintsvp;
+    SV *hintsv;
 
-    if (SORTHINTS(hintsvp) & HINT_SORT_FAST)
-        S_qsortsvu(aTHX_ list1, nmemb, cmp);
-    else {
+    if (SORTHINTS(hintsv) & HINT_SORT_STABLE) {
         register gptr **pp, *q;
         register size_t n, j, i;
         gptr *small[SMALLSORT], **indir, tmp;
@@ -1191,11 +1363,12 @@ S_qsortsv(pTHX_ gptr *list1, size_t nmemb, SVCOMPARE_t cmp)
         /* Copy pointers to original array elements into indirect array */
         for (n = nmemb, pp = indir, q = list1; n--; ) *pp++ = q++;
 
-        savecmp = RealCmp;     /* Save current comparison routine, if any */
-        RealCmp = cmp; /* Put comparison routine where cmpindir can find it */
+        savecmp = PL_sort_RealCmp;     /* Save current comparison routine, if any */
+        PL_sort_RealCmp = cmp; /* Put comparison routine where cmpindir can find it */
 
         /* sort, with indirection */
-        S_qsortsvu(aTHX_ (gptr *)indir, nmemb, cmpindir);
+        S_qsortsvu(aTHX_ (gptr *)indir, nmemb,
+                   flags ? cmpindir_desc : cmpindir);
 
         pp = indir;
         q = list1;
@@ -1237,48 +1410,90 @@ S_qsortsv(pTHX_ gptr *list1, size_t nmemb, SVCOMPARE_t cmp)
        /* free iff allocated */
         if (indir != small) { Safefree(indir); }
         /* restore prevailing comparison routine */
-        RealCmp = savecmp;
+        PL_sort_RealCmp = savecmp;
+    } else if (flags) {
+        SVCOMPARE_t savecmp = PL_sort_RealCmp; /* Save current comparison routine, if any */
+        PL_sort_RealCmp = cmp; /* Put comparison routine where cmp_desc can find it */
+        cmp = cmp_desc;
+        S_qsortsvu(aTHX_ list1, nmemb, cmp);
+        /* restore prevailing comparison routine */
+        PL_sort_RealCmp = savecmp;
+    } else {
+        S_qsortsvu(aTHX_ list1, nmemb, cmp);
     }
 }
 
 /*
+=head1 Array Manipulation Functions
+
 =for apidoc sortsv
 
 Sort an array. Here is an example:
 
     sortsv(AvARRAY(av), av_len(av)+1, Perl_sv_cmp_locale);
 
+See lib/sort.pm for details about controlling the sorting algorithm.
+
 =cut
 */
 
 void
 Perl_sortsv(pTHX_ SV **array, size_t nmemb, SVCOMPARE_t cmp)
 {
-    void (*sortsvp)(pTHX_ SV **array, size_t nmemb, SVCOMPARE_t cmp) =
-        S_mergesortsv;
-    SV **hintsvp;
+    void (*sortsvp)(pTHX_ SV **array, size_t nmemb, SVCOMPARE_t cmp, U32 flags)
+      = S_mergesortsv;
+    SV *hintsv;
     I32 hints;
 
-    if ((hints = SORTHINTS(hintsvp))) {
-        if (hints & HINT_SORT_QUICKSORT)
-             sortsvp = S_qsortsv;
-        else {
-             if (hints & HINT_SORT_MERGESORT)
-                  sortsvp = S_mergesortsv;
-             else
-                  sortsvp = S_mergesortsv;
-        }
+    /*  Sun's Compiler (cc: WorkShop Compilers 4.2 30 Oct 1996 C 4.2) used 
+       to miscompile this function under optimization -O.  If you get test 
+       errors related to picking the correct sort() function, try recompiling 
+       this file without optimiziation.  -- A.D.  4/2002.
+    */
+    hints = SORTHINTS(hintsv);
+    if (hints & HINT_SORT_QUICKSORT) {
+       sortsvp = S_qsortsv;
+    }
+    else {
+       /* The default as of 5.8.0 is mergesort */
+       sortsvp = S_mergesortsv;
+    }
+
+    sortsvp(aTHX_ array, nmemb, cmp, 0);
+}
+
+
+void
+S_sortsv_desc(pTHX_ SV **array, size_t nmemb, SVCOMPARE_t cmp)
+{
+    void (*sortsvp)(pTHX_ SV **array, size_t nmemb, SVCOMPARE_t cmp, U32 flags)
+      = S_mergesortsv;
+    SV *hintsv;
+    I32 hints;
+
+    /*  Sun's Compiler (cc: WorkShop Compilers 4.2 30 Oct 1996 C 4.2) used 
+       to miscompile this function under optimization -O.  If you get test 
+       errors related to picking the correct sort() function, try recompiling 
+       this file without optimiziation.  -- A.D.  4/2002.
+    */
+    hints = SORTHINTS(hintsv);
+    if (hints & HINT_SORT_QUICKSORT) {
+       sortsvp = S_qsortsv;
+    }
+    else {
+       /* The default as of 5.8.0 is mergesort */
+       sortsvp = S_mergesortsv;
     }
 
-    sortsvp(aTHX_ array, nmemb, cmp);
+    sortsvp(aTHX_ array, nmemb, cmp, 1);
 }
 
 PP(pp_sort)
 {
     dSP; dMARK; dORIGMARK;
-    register SV **up;
-    SV **myorigmark = ORIGMARK;
-    register I32 max;
+    register SV **p1 = ORIGMARK+1, **p2;
+    register I32 max, i;
+    AV* av = Nullav;
     HV *stash;
     GV *gv;
     CV *cv = 0;
@@ -1287,6 +1502,11 @@ PP(pp_sort)
     I32 overloading = 0;
     bool hasargs = FALSE;
     I32 is_xsub = 0;
+    I32 sorting_av = 0;
+    U8 private = PL_op->op_private;
+    U8 flags = PL_op->op_flags;
+    void (*sortsvp)(pTHX_ SV **array, size_t nmemb, SVCOMPARE_t cmp)
+      = Perl_sortsv;
 
     if (gimme != G_ARRAY) {
        SP = MARK;
@@ -1295,8 +1515,8 @@ PP(pp_sort)
 
     ENTER;
     SAVEVPTR(PL_sortcop);
-    if (PL_op->op_flags & OPf_STACKED) {
-       if (PL_op->op_flags & OPf_SPECIAL) {
+    if (flags & OPf_STACKED) {
+       if (flags & OPf_SPECIAL) {
            OP *kid = cLISTOP->op_first->op_sibling;    /* pass pushmark */
            kid = kUNOP->op_first;                      /* pass rv2gv */
            kid = kUNOP->op_first;                      /* pass leave */
@@ -1319,8 +1539,8 @@ PP(pp_sort)
                else if (gv) {
                    SV *tmpstr = sv_newmortal();
                    gv_efullname3(tmpstr, gv, Nullch);
-                   DIE(aTHX_ "Undefined sort subroutine \"%s\" called",
-                       SvPVX(tmpstr));
+                   DIE(aTHX_ "Undefined sort subroutine \"%"SVf"\" called",
+                       tmpstr);
                }
                else {
                    DIE(aTHX_ "Undefined subroutine in sort");
@@ -1334,8 +1554,7 @@ PP(pp_sort)
                SAVEVPTR(CvROOT(cv)->op_ppaddr);
                CvROOT(cv)->op_ppaddr = PL_ppaddr[OP_NULL];
 
-               SAVEVPTR(PL_curpad);
-               PL_curpad = AvARRAY((AV*)AvARRAY(CvPADLIST(cv))[1]);
+               PAD_SET_CUR(CvPADLIST(cv), 1);
             }
        }
     }
@@ -1344,24 +1563,59 @@ PP(pp_sort)
        stash = CopSTASH(PL_curcop);
     }
 
-    up = myorigmark + 1;
-    while (MARK < SP) {        /* This may or may not shift down one here. */
-       /*SUPPRESS 560*/
-       if ((*up = *++MARK)) {                  /* Weed out nulls. */
-           SvTEMP_off(*up);
-           if (!PL_sortcop && !SvPOK(*up)) {
+    /* optimiser converts "@a = sort @a" to "sort \@a";
+     * in case of tied @a, pessimise: push (@a) onto stack, then assign
+     * result back to @a at the end of this function */
+    if (private & OPpSORT_INPLACE) {
+       assert( MARK+1 == SP && *SP && SvTYPE(*SP) == SVt_PVAV);
+       (void)POPMARK; /* remove mark associated with ex-OP_AASSIGN */
+       av = (AV*)(*SP);
+       max = AvFILL(av) + 1;
+       if (SvMAGICAL(av)) {
+           MEXTEND(SP, max);
+           p2 = SP;
+           for (i=0; i < (U32)max; i++) {
+               SV **svp = av_fetch(av, i, FALSE);
+               *SP++ = (svp) ? *svp : Nullsv;
+           }
+       }
+       else {
+           p1 = p2 = AvARRAY(av);
+           sorting_av = 1;
+       }
+    }
+    else {
+       p2 = MARK+1;
+       max = SP - MARK;
+   }
+
+    if (private & OPpSORT_DESCEND) {
+       sortsvp = S_sortsv_desc;
+    }
+
+    /* shuffle stack down, removing optional initial cv (p1!=p2), plus any
+     * nulls; also stringify any args */
+    for (i=max; i > 0 ; i--) {
+       if ((*p1 = *p2++)) {                    /* Weed out nulls. */
+           SvTEMP_off(*p1);
+           if (!PL_sortcop && !SvPOK(*p1)) {
                STRLEN n_a;
-               if (SvAMAGIC(*up))
+               if (SvAMAGIC(*p1))
                    overloading = 1;
                else
-                   (void)sv_2pv(*up, &n_a);
+                   (void)sv_2pv(*p1, &n_a);
            }
-           up++;
+           p1++;
        }
+       else
+           max--;
     }
-    max = --up - myorigmark;
-    if (PL_sortcop) {
-       if (max > 1) {
+    if (sorting_av)
+       AvFILLp(av) = max-1;
+
+    if (max > 1) {
+       SV **start;
+       if (PL_sortcop) {
            PERL_CONTEXT *cx;
            SV** newsp;
            bool oldcatch = CATCH_GET;
@@ -1379,50 +1633,43 @@ PP(pp_sort)
                    PL_secondgv = gv_fetchpv("b", TRUE, SVt_PV);
                    PL_sortstash = stash;
                }
-#ifdef USE_5005THREADS
-               sv_lock((SV *)PL_firstgv);
-               sv_lock((SV *)PL_secondgv);
-#endif
                SAVESPTR(GvSV(PL_firstgv));
                SAVESPTR(GvSV(PL_secondgv));
            }
 
            PUSHBLOCK(cx, CXt_NULL, PL_stack_base);
-           if (!(PL_op->op_flags & OPf_SPECIAL)) {
+           if (!(flags & OPf_SPECIAL)) {
                cx->cx_type = CXt_SUB;
                cx->blk_gimme = G_SCALAR;
                PUSHSUB(cx);
-               if (!CvDEPTH(cv))
-                   (void)SvREFCNT_inc(cv); /* in preparation for POPSUB */
            }
            PL_sortcxix = cxstack_ix;
 
            if (hasargs && !is_xsub) {
                /* This is mostly copied from pp_entersub */
-               AV *av = (AV*)PL_curpad[0];
+               AV *av = (AV*)PAD_SVl(0);
 
-#ifndef USE_5005THREADS
                cx->blk_sub.savearray = GvAV(PL_defgv);
                GvAV(PL_defgv) = (AV*)SvREFCNT_inc(av);
-#endif /* USE_5005THREADS */
-               cx->blk_sub.oldcurpad = PL_curpad;
+               CX_CURPAD_SAVE(cx->blk_sub);
                cx->blk_sub.argarray = av;
            }
-           sortsv((myorigmark+1), max,
-                  is_xsub ? sortcv_xsub : hasargs ? sortcv_stacked : sortcv);
+           
+           start = p1 - max;
+           sortsvp(aTHX_ start, max,
+                   is_xsub ? sortcv_xsub : hasargs ? sortcv_stacked : sortcv);
 
            POPBLOCK(cx,PL_curpm);
            PL_stack_sp = newsp;
            POPSTACK;
            CATCH_SET(oldcatch);
        }
-    }
-    else {
-       if (max > 1) {
+       else {
            MEXTEND(SP, 20);    /* Can't afford stack realloc on signal. */
-           sortsv(ORIGMARK+1, max,
-                  (PL_op->op_private & OPpSORT_NUMERIC)
-                       ? ( (PL_op->op_private & OPpSORT_INTEGER)
+           start = sorting_av ? AvARRAY(av) : ORIGMARK+1;
+           sortsvp(aTHX_ start, max,
+                   (private & OPpSORT_NUMERIC)
+                       ? ( (private & OPpSORT_INTEGER)
                            ? ( overloading ? amagic_i_ncmp : sv_i_ncmp)
                            : ( overloading ? amagic_ncmp : sv_ncmp))
                        : ( IN_LOCALE_RUNTIME
@@ -1430,19 +1677,38 @@ PP(pp_sort)
                                ? amagic_cmp_locale
                                : sv_cmp_locale_static)
                            : ( overloading ? amagic_cmp : sv_cmp_static)));
-           if (PL_op->op_private & OPpSORT_REVERSE) {
-               SV **p = ORIGMARK+1;
-               SV **q = ORIGMARK+max;
-               while (p < q) {
-                   SV *tmp = *p;
-                   *p++ = *q;
-                   *q-- = tmp;
-               }
+       }
+       if (private & OPpSORT_REVERSE) {
+           SV **q = start+max-1;
+           while (start < q) {
+               SV *tmp = *start;
+               *start++ = *q;
+               *q-- = tmp;
            }
        }
     }
+    if (av && !sorting_av) {
+       /* simulate pp_aassign of tied AV */
+       SV *sv;
+       SV** base, **didstore;
+       for (base = ORIGMARK+1, i=0; i < max; i++) {
+           sv = NEWSV(28,0);
+           sv_setsv(sv, base[i]);
+           base[i] = sv;
+       }
+       av_clear(av);
+       av_extend(av, max);
+       for (i=0; i < max; i++) {
+           sv = base[i];
+           didstore = av_store(av, i, sv);
+           if (SvSMAGICAL(sv))
+               mg_set(sv);
+           if (!didstore)
+               sv_2mortal(sv);
+       }
+    }
     LEAVE;
-    PL_stack_sp = ORIGMARK + max;
+    PL_stack_sp = ORIGMARK + (sorting_av ? 0 : max);
     return nextop;
 }
 
@@ -1477,11 +1743,7 @@ sortcv_stacked(pTHX_ SV *a, SV *b)
     I32 result;
     AV *av;
 
-#ifdef USE_5005THREADS
-    av = (AV*)PL_curpad[0];
-#else
     av = GvAV(PL_defgv);
-#endif
 
     if (AvMAX(av) < 1) {
        SV** ary = AvALLOC(av);
@@ -1656,5 +1918,3 @@ amagic_cmp_locale(pTHX_ register SV *str1, register SV *str2)
     }
     return sv_cmp_locale(str1, str2);
 }
-
-