Merge pull request #207 from dicej/composable-continuations

Composable continuations

classpath/avian/Cell.java

@@ -31,4 +31,24 @@ public class Cell <T> {
     sb.append(")");
     return sb.toString();
   }
+
+  public static <Car> Cell<Car> cons(Car car, Cell<Car> cdr) {
+    return new Cell(car, cdr);
+  }
+
+  public static <T> boolean equal(T a, T b) {
+    return (a == null && b == null) || (a != null && a.equals(b));
+  }
+  
+  public static <Car> boolean equal(Cell<Car> a, Cell<Car> b) {
+    while (a != null) {
+      if (b == null || (! equal(a.value, b.value))) {
+        return false;
+      }
+      a = a.next;
+      b = b.next;
+    }
+
+    return b == null;
+  }
 }

classpath/avian/Continuations.java

@@ -120,19 +120,21 @@ import java.util.concurrent.Callable;
 public class Continuations {
   private Continuations() { }
 
+  private static final ThreadLocal<Reset> latestReset = new ThreadLocal();
+
   /**
    * Captures the current continuation, passing a reference to the
    * specified receiver.
    *
    * <p>This method will either return the result returned by
-   * <code>receiver.receive(Callback)</code>, propagate the exception
+   * <code>receiver.call(Callback)</code>, propagate the exception
    * thrown by that method, return the result passed to the
    * handleResult(T) method of the continuation, or throw the
    * exception passed to the handleException(Throwable) method of the
    * continuation.
    */
   public static native <T> T callWithCurrentContinuation
-    (CallbackReceiver<T> receiver) throws Exception;
+    (Function<Callback<T>,T> receiver) throws Exception;
 
   /**
    * Calls the specified "before" and "after" tasks each time a
@@ -181,6 +183,83 @@ public class Continuations {
     }
   }
 
+  public static <B,C> C reset(final Callable<B> thunk) throws Exception {
+    final Reset reset = new Reset(latestReset.get());
+    latestReset.set(reset);
+    try {
+      Object result = callWithCurrentContinuation
+        (new Function<Callback<Object>,Object>() {
+          public Object call(Callback continuation) throws Exception {
+            reset.continuation = continuation;
+            return thunk.call();
+          }
+        });
+
+      while (true) {
+        Cell<Function> shift = reset.shifts;
+        if (shift != null) {
+          reset.shifts = shift.next;
+          result = shift.value.call(result);
+        } else {
+          return (C) result;
+        }
+      }
+    } finally {
+      latestReset.set(reset.next);
+    }
+  }
+
+  public static <A,B,C> A shift
+    (final Function<Function<A,B>,C> receiver)
+    throws Exception
+  {
+    return (A) callWithCurrentContinuation
+      (new Function<Callback<Object>,Object>() {
+        public Object call(final Callback continuation) {
+          final Reset reset = latestReset.get();
+          reset.shifts = new Cell(new Function() {
+              public Object call(Object ignored) throws Exception {
+                return receiver.call
+                  (new Function() {
+                      public Object call(final Object argument)
+                        throws Exception
+                      {
+                        return callWithCurrentContinuation
+                          (new Function<Callback<Object>,Object>() {
+                            public Object call
+                              (final Callback shiftContinuation)
+                              throws Exception
+                            {
+                              reset.shifts = new Cell
+                                (new Function() {
+                                    public Object call(Object result)
+                                      throws Exception
+                                    {
+                                      shiftContinuation.handleResult(result);
+                                      throw new AssertionError();
+                                    }
+                                  },
+                                  reset.shifts);
+
+                              continuation.handleResult(argument);
+                              throw new AssertionError();
+                            }
+                          });
+                      }
+                    });
+              }
+
+              public void handleException(Throwable exception) {
+                throw new AssertionError();
+              }
+            }, reset.shifts);
+
+          reset.continuation.handleResult(null);
+          throw new AssertionError();
+        }
+      });
+  }
+
   private static native UnwindResult dynamicWind2(Runnable before,
                                                   Callable thunk,
                                                   Runnable after)
@@ -235,4 +314,14 @@ public class Continuations {
       this.exception = exception;
     }
   }
+
+  private static class Reset {
+    public Callback continuation;
+    public final Reset next;
+    public Cell<Function> shifts;
+
+    public Reset(Reset next) {
+      this.next = next;
+    }
+  }
 }

classpath/avian/Function.java

@@ -10,6 +10,6 @@
 
 package avian;
 
-public interface CallbackReceiver<T> {
-  public T receive(Callback<T> callback) throws Exception;
+public interface Function<A,B> {
+  public B call(A argument) throws Exception;
 }

makefile

@@ -192,7 +192,7 @@ ifneq ($(android),)
 	crypto-native := $(android)/libcore/crypto/src/main/native
 
 	ifeq ($(platform),windows)
-    	crypto-cpps := $(crypto-native)/org_conscrypt_NativeCrypto.cpp
+		crypto-cpps := $(crypto-native)/org_conscrypt_NativeCrypto.cpp
 		android-cflags += -D__STDC_CONSTANT_MACROS
 		blacklist = $(luni-native)/java_io_Console.cpp \
 			$(luni-native)/java_lang_ProcessManager.cpp \
@@ -1330,13 +1330,13 @@ ifneq ($(classpath),avian)
 		$(classpath-src)/avian/AnnotationInvocationHandler.java \
 		$(classpath-src)/avian/Assembler.java \
 		$(classpath-src)/avian/Callback.java \
-		$(classpath-src)/avian/CallbackReceiver.java \
 		$(classpath-src)/avian/ClassAddendum.java \
 		$(classpath-src)/avian/InnerClassReference.java \
 		$(classpath-src)/avian/Classes.java \
 		$(classpath-src)/avian/ConstantPool.java \
 		$(classpath-src)/avian/Continuations.java \
 		$(classpath-src)/avian/FieldAddendum.java \
+		$(classpath-src)/avian/Function.java \
 		$(classpath-src)/avian/IncompatibleContinuationException.java \
 		$(classpath-src)/avian/Machine.java \
 		$(classpath-src)/avian/MethodAddendum.java \
@@ -1397,6 +1397,7 @@ unittest-depends = \
 
 ifeq ($(continuations),true)
 	continuation-tests = \
+		extra.ComposableContinuations \
 		extra.Continuations \
 		extra.Coroutines \
 		extra.DynamicWind

src/compile.cpp

@@ -8044,8 +8044,8 @@ callWithCurrentContinuation(MyThread* t, object receiver)
 
     if (root(t, ReceiveMethod) == 0) {
       object m = resolveMethod
-        (t, root(t, Machine::BootLoader), "avian/CallbackReceiver", "receive",
-         "(Lavian/Callback;)Ljava/lang/Object;");
+        (t, root(t, Machine::BootLoader), "avian/Function", "call",
+         "(Ljava/lang/Object;)Ljava/lang/Object;");
 
       if (m) {
         setRoot(t, ReceiveMethod, m);

src/types.def

@@ -176,8 +176,6 @@
 
 (type unwindResult avian/Continuations$UnwindResult)
 
-(type callbackReceiver avian/CallbackReceiver)
-
 (type string java/lang/String
   (alias data object value)
   (alias length uint32_t count)

test/extra/ComposableContinuations.java

@@ -0,0 +1,93 @@
+package extra;
+
+import static avian.Continuations.shift;
+import static avian.Cell.cons;
+import static avian.Cell.equal;
+
+import avian.Cell;
+import avian.Function;
+import avian.Continuations;
+
+import java.util.concurrent.Callable;
+
+public class ComposableContinuations {
+  private static void expect(boolean v) {
+    if (! v) throw new RuntimeException();
+  }
+
+  public static void main(String[] args) throws Exception {
+    expect(2 * Continuations.<Integer,Integer>reset(new Callable<Integer>() {
+          public Integer call() throws Exception {
+            return 1 + shift
+              (new Function<Function<Integer,Integer>,Integer>() {
+                public Integer call(Function<Integer,Integer> continuation)
+                  throws Exception
+                {
+                  return continuation.call(5);
+                }
+              });
+          }
+        }) == 12);
+
+    expect(1 + Continuations.<Integer,Integer>reset(new Callable<Integer>() {
+          public Integer call() throws Exception {
+            return 2 * shift
+              (new Function<Function<Integer,Integer>,Integer>() {
+                public Integer call(Function<Integer,Integer> continuation)
+                  throws Exception
+                {
+                  return continuation.call(continuation.call(4));
+                }
+              });
+          }
+        }) == 17);
+
+    expect
+      (equal
+       (Continuations.<Cell<Integer>,Cell<Integer>>reset
+        (new Callable<Cell<Integer>>() {
+          public Cell<Integer> call() throws Exception {
+            shift(new Function<Function<Cell<Integer>,Cell<Integer>>,
+                  Cell<Integer>>()
+                  {
+                    public Cell<Integer> call
+                      (Function<Cell<Integer>,Cell<Integer>> continuation)
+                      throws Exception
+                    {
+                      return cons(1, continuation.call(null));
+                    }
+                  });
+
+            shift(new Function<Function<Cell<Integer>,Cell<Integer>>,
+                  Cell<Integer>>()
+                  {
+                    public Cell<Integer> call
+                      (Function<Cell<Integer>,Cell<Integer>> continuation)
+                      throws Exception
+                    {
+                      return cons(2, continuation.call(null));
+                    }
+                  });
+
+            return null;
+          }
+        }), cons(1, cons(2, null))));
+
+    expect
+      (equal
+       (Continuations.<String,Integer>reset
+        (new Callable<String>() {
+          public String call() throws Exception {
+            return new String
+              (shift(new Function<Function<byte[],String>,Integer>() {
+                  public Integer call(Function<byte[],String> continuation)
+                    throws Exception
+                  {
+                    return Integer.parseInt
+                      (continuation.call(new byte[] { 0x34, 0x32 }));
+                  }
+                }), "UTF-8");
+          }
+        }), 42));
+  }
+}

test/extra/Continuations.java

@@ -2,7 +2,7 @@ package extra;
 
 import static avian.Continuations.callWithCurrentContinuation;
 
-import avian.CallbackReceiver;
+import avian.Function;
 import avian.Callback;
 
 public class Continuations {
@@ -11,22 +11,26 @@ public class Continuations {
   }
 
   public static void main(String[] args) throws Exception {
-    expect(callWithCurrentContinuation(new CallbackReceiver<Integer>() {
-          public Integer receive(Callback<Integer> continuation) {
-            continuation.handleResult(42);
-            throw new AssertionError();
-          }
-        }) == 42);
+    expect
+      (callWithCurrentContinuation
+       (new Function<Callback<Integer>,Integer>() {
+         public Integer call(Callback<Integer> continuation) {
+           continuation.handleResult(42);
+           throw new AssertionError();
+         }
+       }) == 42);
 
-    expect(callWithCurrentContinuation(new CallbackReceiver<Integer>() {
-          public Integer receive(Callback<Integer> continuation) {
-            return 43;
-          }
-        }) == 43);
+    expect
+      (callWithCurrentContinuation
+       (new Function<Callback<Integer>,Integer>() {
+         public Integer call(Callback<Integer> continuation) {
+           return 43;
+         }
+       }) == 43);
 
     try {
-      callWithCurrentContinuation(new CallbackReceiver<Integer>() {
-          public Integer receive(Callback<Integer> continuation) {
+      callWithCurrentContinuation(new Function<Callback<Integer>,Integer>() {
+          public Integer call(Callback<Integer> continuation) {
             continuation.handleException(new MyException());
             throw new AssertionError();
           }
@@ -37,8 +41,8 @@ public class Continuations {
     }
 
     try {
-      callWithCurrentContinuation(new CallbackReceiver<Integer>() {
-          public Integer receive(Callback<Integer> continuation)
+      callWithCurrentContinuation(new Function<Callback<Integer>,Integer>() {
+          public Integer call(Callback<Integer> continuation)
             throws MyException
           {
             throw new MyException();

test/extra/Coroutines.java

@@ -2,7 +2,7 @@ package extra;
 
 import static avian.Continuations.callWithCurrentContinuation;
 
-import avian.CallbackReceiver;
+import avian.Function;
 import avian.Callback;
 
 public class Coroutines {
@@ -40,8 +40,8 @@ public class Coroutines {
 
     final Consumer<Character> consumer = new Consumer<Character>() {
       public void consume(final Character c) throws Exception {
-        callWithCurrentContinuation(new CallbackReceiver() {
-          public Object receive(Callback continuation) {
+        callWithCurrentContinuation(new Function<Callback<Object>,Object>() {
+          public Object call(Callback continuation) {
             state.produceNext = continuation;
 
             state.consumeNext.handleResult(c);
@@ -53,9 +53,9 @@ public class Coroutines {
     };
 
     final Producer<Character> producer = new Producer<Character>() {
-      final CallbackReceiver<Character> receiver
-      = new CallbackReceiver<Character>() {
-        public Character receive(Callback<Character> continuation)
+      final Function<Callback<Character>,Character> receiver
+      = new Function<Callback<Character>,Character>() {
+        public Character call(Callback<Character> continuation)
         throws Exception
         {
           state.consumeNext = continuation;

test/extra/DynamicWind.java

@@ -3,7 +3,7 @@ package extra;
 import static avian.Continuations.callWithCurrentContinuation;
 import static avian.Continuations.dynamicWind;
 
-import avian.CallbackReceiver;
+import avian.Function;
 import avian.Callback;
 
 import java.util.concurrent.Callable;
@@ -86,24 +86,27 @@ public class DynamicWind {
       });
   }
 
-  private void continuationUnwindTest(final CallbackReceiver<Integer> receiver)
+  private void continuationUnwindTest
+    (final Function<Callback<Integer>,Integer> receiver)
     throws Exception
   {
     System.out.println("continuationUnwindTest enter");
             
     try {
-      expect(callWithCurrentContinuation(new CallbackReceiver<Integer>() {
-            public Integer receive(final Callback<Integer> continuation)
-              throws Exception
-            {
-              unwindTest(new Callable<Integer>() {
-                  public Integer call() throws Exception {
-                    return receiver.receive(continuation);
-                  }
-                });
-              throw new AssertionError();
-            }
-          }) == 42);
+      expect
+        (callWithCurrentContinuation
+         (new Function<Callback<Integer>,Integer>() {
+           public Integer call(final Callback<Integer> continuation)
+             throws Exception
+           {
+             unwindTest(new Callable<Integer>() {
+                 public Integer call() throws Exception {
+                   return receiver.call(continuation);
+                 }
+               });
+             throw new AssertionError();
+           }
+         }) == 42);
     } catch (MyException e) {
       e.printStackTrace();
     }
@@ -118,8 +121,8 @@ public class DynamicWind {
   }
 
   private void continuationResultUnwind() throws Exception {
-    continuationUnwindTest(new CallbackReceiver<Integer>() {
-          public Integer receive(final Callback<Integer> continuation) {
+    continuationUnwindTest(new Function<Callback<Integer>,Integer>() {
+          public Integer call(final Callback<Integer> continuation) {
             continuation.handleResult(42);
             throw new AssertionError();
           }
@@ -127,8 +130,8 @@ public class DynamicWind {
   }
 
   private void continuationExceptionUnwind() throws Exception {
-    continuationUnwindTest(new CallbackReceiver<Integer>() {
-          public Integer receive(final Callback<Integer> continuation) {
+    continuationUnwindTest(new Function<Callback<Integer>,Integer>() {
+          public Integer call(final Callback<Integer> continuation) {
             continuation.handleException(new MyException());
             throw new AssertionError();
           }
@@ -163,8 +166,8 @@ public class DynamicWind {
             task = 1;
         
             return callWithCurrentContinuation
-            (new CallbackReceiver<Integer>() {
-              public Integer receive(final Callback<Integer> continuation)
+            (new Function<Callback<Integer>,Integer>() {
+              public Integer call(final Callback<Integer> continuation)
                 throws Exception
               {
                 continuationReference = continuation;

test/test.sh

@@ -18,7 +18,7 @@ fi
 
 echo -n "" >${log}
 
-printf "%12s------- Unit tests -------\n" ""
+printf "%20s------- Unit tests -------\n" ""
 ${unit_tester} 2>>${log}
 if [ "${?}" != "0" ]; then
   trouble=1
@@ -27,9 +27,9 @@ fi
 
 echo
 
-printf "%12s------- Java tests -------\n" ""
+printf "%20s------- Java tests -------\n" ""
 for test in ${tests}; do
-  printf "%24s: " "${test}"
+  printf "%32s: " "${test}"
 
   case ${mode} in
     debug|debug-fast|fast|small )

unittest/test-harness.cpp

@@ -32,7 +32,7 @@ Test::Test(const char* name):
 bool Test::runAll() {
   int failures = 0;
   for(Test* t = Test::first; t; t = t->next) {
-    printf("%24s: ", t->name);
+    printf("%32s: ", t->name);
     t->run();
     failures += t->failures;
     if(t->failures > 0) {
@@ -49,4 +49,4 @@ int main(int argc UNUSED, char** argv UNUSED) {
     return 0;
   }
   return 1;
-}
+}