fix cilksort segfault/slowness

Author: michaelborkowski

src/CilkSort.hs

@@ -43,37 +43,56 @@ cilkSortInplace :: (Show a, HasPrimOrd a, NFData a) =>
 cilkSortInplace :: (Show a, HasPrimOrd a) =>
 #endif
   A.Array a -. A.Array a -. (A.Array a, A.Array a)
-cilkSortInplace src tmp =
-  let !(Ur len, src') = A.size2 src in
-  if len <= SEQSIZE
-  then
-    if len <= QUICKSIZE
-    then let src'' = quickSort src'
-          in (src'', tmp)
-    else Seq.msortInplace src' tmp
-  else
-    let !(srcA, srcB)     = splitMid src'
-        !(tmpA, tmpB)     = splitMid tmp
-        !(src1, src2)     = splitMid srcA
-        !(src3, src4)     = splitMid srcB
-        !(tmp1, tmp2)     = splitMid tmpA
-        !(tmp3, tmp4)     = splitMid tmpB
-        !(((src1', tmp1'), (src2', tmp2')), ((src3', tmp3'), (src4', tmp4')))
-                         = (.||||.) (cilkSortInplace src1 tmp1) (cilkSortInplace src2 tmp2)
-                                    (cilkSortInplace src3 tmp3) (cilkSortInplace src4 tmp4)
-        tmpA'            = A.append tmp1' tmp2'
-        tmpB'            = A.append tmp3' tmp4'
-        !((srcA'', tmpA''), (srcB'', tmpB''))
-                         = merge_par src1' src2' tmpA' .||. merge_par src3' src4' tmpB'
---                         = tuple2 (merge_par src1' src2') tmpA' (merge_par src3' src4') tmpB'
-        src''            = A.append srcA'' srcB''
-        !(tmp''', src''') = merge_par tmpA'' tmpB'' src''
-    in  (src''', tmp''') ? lem_toBag_splitMid src
-                         ? lem_toBag_splitMid tmp
-                         ? lem_toBag_splitMid srcA
-                         ? lem_toBag_splitMid srcB
-                         ? lem_toBag_splitMid tmpA
-                         ? lem_toBag_splitMid tmpB
+cilkSortInplace src tmp = go src tmp where
+  {-@ go ::xs:Array a
+      -> { ys:(Array a ) | A.size ys  == A.size xs   && left xs == left ys &&
+                           right xs == right ys }
+      -> (Array a, Array a)<{\zs ts -> toBag xs == toBag zs && isSorted' zs &&
+                                       token xs == token zs && token ys == token ts &&
+                                       A.size xs == A.size zs && A.size ys == A.size ts &&
+                                       left zs == left xs && right zs == right xs &&
+                                       left ts == left ys && right ts == right ys }>
+       / [A.size xs] @-}
+#ifdef MUTABLE_ARRAYS
+  go :: (Show a, HasPrimOrd a, NFData a) =>
+#else
+  go :: (Show a, HasPrimOrd a) =>
+#endif
+    A.Array a -. A.Array a -. (A.Array a, A.Array a)
+  go src tmp = 
+    let !(Ur len, src') = A.size2 src in
+    if len <= SEQSIZE
+    then
+      if len <= QUICKSIZE
+      then let src'' = quickSort' src'
+            in (src'', tmp)
+      else Seq.msortInplace src' tmp
+    else
+      let !(srcA, srcB)     = splitMid src'
+          !(tmpA, tmpB)     = splitMid tmp
+          !(src1, src2)     = splitMid srcA
+          !(src3, src4)     = splitMid srcB
+          !(tmp1, tmp2)     = splitMid tmpA
+          !(tmp3, tmp4)     = splitMid tmpB
+          !(((src1', tmp1'), (src2', tmp2')), ((src3', tmp3'), (src4', tmp4')))
+                           = (go src1 tmp1 .||. go src2 tmp2) .||.
+                             (go src3 tmp3 .||. go src4 tmp4)
+  --                         = (.||||.) (go src1 tmp1) (go src2 tmp2)
+  --                                    (go src3 tmp3) (go src4 tmp4)
+          tmpA'            = A.append tmp1' tmp2'
+          tmpB'            = A.append tmp3' tmp4'
+          !((srcA'', tmpA''), (srcB'', tmpB''))
+                           = merge_par src1' src2' tmpA' .||. merge_par src3' src4' tmpB'
+    --                         = tuple2 (merge_par src1' src2') tmpA' (merge_par src3' src4') tmpB'
+          src''            = A.append srcA'' srcB''
+          !(tmp''', src''') = merge_par tmpA'' tmpB'' src''
+       in  (src''', tmp''') ? lem_toBag_splitMid src
+                            ? lem_toBag_splitMid tmp
+                            ? lem_toBag_splitMid srcA
+                            ? lem_toBag_splitMid srcB
+                            ? lem_toBag_splitMid tmpA
+                            ? lem_toBag_splitMid tmpB
+{-# INLINE cilkSortInplace #-}                            
 
 {-@ cilkSort' :: y:a
            -> { xs:(Array a) | A.size xs > 0 && left xs == 0 && right xs == size xs && y == A.get xs 0 }
@@ -89,6 +108,7 @@ cilkSort' anyVal src =
   let !(Ur len, src') = A.size2 src
       !src'' = A.allocScratch len anyVal cilkSortInplace src' in
   src''
+{-# INLINE cilkSort' #-}  
 
 -- finally, the top-level merge sort function
 {-@ cilkSort :: { xs:(A.Array a) | left xs == 0 && right xs == size xs }
@@ -104,3 +124,4 @@ cilkSort src =
   let !(Ur len, src') = A.size2 src in
       if len == 0 then src'
       else let !(Ur x0, src'') = A.get2 0 src' in cilkSort' x0 src''
+{-# INLINABLE cilkSort #-}