翻译-QCon London 2019-Why Continua

QCon London 2019- Why Continuations are Coming to Java

• talks说明

  • 讨论并比较了处理和并发的各种技术如纯函数式（monads, affine types）和命令式编程（threads, continuations, monads, async/await）
  • 说明了为什么continuations是非常适合命令式风格

• talks笔记

  • Why Java is Getting Continuations

  • 关于计算的观点

    • 确定性的、顺序的
    • 不确定性的、交互的/并发的

  • 纯函数式编程

    • 确定性、顺序的

    • Linear Types: nondeterminism as a function of unknown values

      • 线性类型-未知函数的不确定性

    • IO Type: Move nondeterminism outside the program

      • io类型-将不确定性转移到程序之外

    • landon: 这里主要是函数式编程的一些知识，如haskell、monad等

  • 经典的命令式编程

    • 不确定性的、并发的/交互的

  • 反应式编程的问题

    // 一个阻塞方法
    double calcImportantFinance(double x) {
        try { 
            double result; 
            result = compute("USD->euro", x); 
            result = compute("subtractTax", result * 1.3); 
            result = compute("addInterest", result); 
            return result; 
        } catch(Exception ex) { 
            log(ex);  
            throw ex;
        }
    }
    
    double compute(String op, double x);
    
    // 用CompletableFuture异步实现
    CompletableFuture<Double> calcImportantFinance (double x) {
        return
            compute("USD->euro", 100.0)
            .thenCompose(result -> compute("subtractTax", result * 1.3))
            .thenCompose(result -> compute("addInterest", result))
            .handle((result,ex) -> {
                if(ex != null) {
                    log(ex); 
                    throw new RuntimeException(ex);
                }else{
                    return result
                }
            });
    }
    

    • Lost Control Flow

    double result;
    result = compute("USD->euro", x);
    if(result > 1000) {
         result = compute("subtractTax", result * 1.3); 
    }
    while(!sufficient(result)) {
        result = compute("addInterest", result);
    }
    return result;
    

    • CompletableFuture和控制流在某些方面不匹配

    • Lost Context

      • 上面每一步都是在不同的线程执行
      • 当出现异常时，上下文丢失，则后面调试代码非常困难。另外性能分析也会变的困难，因为性能分析是通过对堆栈跟踪采样来进行的。

    • Viral 病毒式的

      • 最大的问题是返回类型，返回值是CompletableFuture<Double>
      • 除非调用阻塞的get方法，否则无法获取值
      • 另外调用链上也必须异步方式使用这些CompletableFuture，这是病毒式的
      • 这意味着我们的这个调用堆栈要么是阻塞同步，要么是异步，两种几乎不能互操作

  • Async/Await

    async Task<double> CalImportantFinace()
    {
        try
        {
            double result; 
            result = await compute("USD->euro", x); 
            result = await compute("subtractTax", result * 1.3); 
            result = await compute("addInterest", result); 
            return result; 
        }
        catch(Exception ex)
        {
            log(ex);
            throw ex;
        }
    }
    

    • C#引入一个叫async and await的东西，功能类似于Java的CompletableFuture，但是可以用await关键字作为异步方法的前缀，看着是比较漂亮的命令式代码
    • 解决了提到的第一个lost control flow的问题。但是对于第二个问题，因为上面的每一行仍然在不同的线程上执行，如果出现异常的话会是一个比较奇怪的异常堆栈
    • .Net Core 2.1尝试修复此问题，它们人为地生成了一个可以捕获实际上下文的堆栈跟踪，但是与您编写普通的阻塞代码时所得到的堆栈跟踪相比，它仍然是非常不同的堆栈跟踪
    • 所以最大的问题还是隔离了同步阻塞世界和异步世界

  • Why give up a good (core!) abstraction just because of an inadequate implementation?

    • 命令式语言不仅处理输入和非确定性语言，还具有线程，进程和阻塞的内置概念，从编程角度来看，它是一个很好的抽象。我们有时想要避免的唯一原因是内核线程的实现负担太重
    • 显而易见的解决方案是更改实施，而不是更改我们的编程方式。这就是我们作为Project Loom的一部分正在尝试做的事情
    • First, we need the ability to stop the code that's running on the CPU and say, "I'm not using the CPU anymore," and later on have the ability to resume it.
      • Continuation，yield and resume
    • we need some mechanism to schedule those pieces that want to run the CPU
      • Scheduler
    • thread/process = yield control and resume + execution scheduling
    • thread/process = continuation + scheduler

  • Continuations

    • 当您调用run时，continuation的主体将运行到完成或直到下一次调用yield为止
    • 作用域使我们可以在另一个内部嵌套不同的continuation，并能够挂起多个continuation并跳回多个调用方，这与我们在堆栈中抛出异常的方式类似
    • 限定的意思是我们要暂停和恢复的代码部分不是整个程序，而只是body内的代码
    • 每次运行它都会一直运行到下一个挂起点，然后状态改变，然后当您再次运行它时，它将从第一个挂起点运行到第二个挂起点，我们永远都无法回到过去
    • 克隆和序列化

  • Fiber

    • 今天，编写应用程序的人可以选择编写简单的同步阻塞代码，但是他们依靠内核为他们提供线程，并且内核只能处理这么多线程。该应用程序将非常清晰，易于维护和调试，但将不可扩展。另一种选择是编写异步代码，正如我向您展示的那样，这要复杂得多，并且很难将现有代码放入其中，但至少它是可伸缩的

    • 使用fiber，如果我们确实可以使它们比内核提供的线程轻得多，则可以解决此问题。您可以根据需要拥有任意数量的这些用户模式轻量级线程，并且可以进行阻塞，阻塞实际上是免费的

    • fiber = continuation + scheduler

    • codes like sync，works like async

      • concurrency made simple

    • 这使编写并发应用程序变得更加容易，因为它可以帮助您匹配业务的并发单元。这是重量级内核线程无法实现的，因为您可能有100,000个并发用户，但是当前线程却无法达到100,000

    • io，java.util.concurrent

      • 目前已经是fiber-blocking了

    • 如果你喜欢Async/Await，那么下面这个就是整个实现

      class Async<T> extends CompletableFuture<T> { 
          public static <T> T await(Async<T> async) throwsInterruptedException, ExecutionException { 
              return async.get(); 
          } 
      }
      

      • 即可以直接在fiber中调用这个await

    • JDK中旧的重量级线程的数据。 它们每个都有大约2 KB的元数据，默认情况下还有1 MB的堆栈。 当涉及到fiber时，它们目前在原型中，它们目前只有2到300字节的元数据，而且可增长和收缩

    • 对于内核管理的重量级线程的任务切换成本，切换任务的时间在1到10微秒之间。 对于fiber，我们不知道它到底有多少，而性能仍然是我们仍在努力的方面，但我们希望它会比现在低得多

  • Rethink threads

    • 结构化并发

      • 线程被限制在一个已知的生命周期内，该生命周期可以扩展到给定的代码块
      • 我们有一个定义fiber作用域的块，保证在退出作用域时终止在作用域内创建的所有fiber
      • fiber作用域可取消，也可以给它一个截止日期

    • 结构化并发

      try(var scope = FiberScope.cancellable()) {
          Fiber<?> fiber = scope.schedule(task)
      }
      

      • 作用域内调度所有fiber都终止之前，无法退出作用域
      • fiber作用域可以嵌套
      • FiberScope具有一个终止队列，用于收集作用域内fiber的结果
      • 取消一个在可取消作用域的fiber会取消所有已经调度的fiber。在这个作用域内，通过嵌套，可以取消一颗fiber树
      • 取消一个park在阻塞io操作的fiber会引起unpark和检查是否取消

    • 结构化并发

      • 返回第一个完整任务的结果，取消并等待所有未完成的fiber终止后再返回

      • fiber作用域为我们提供了终止队列

        <V> V anyOf(Callable<? extends V>[] tasks) throws Throwable { 
            try (var scope = FiberScope.cancellable()) { 
                var queue = new FiberScope.TerminationQueue<V>(); 
                Arrays.stream(tasks).forEach(task -> scope.schedule(task, queue)); 
                
            try { 
                return queue.take().join(); 
             } catch (CompletionException e) { 
                throw e.getCause(); 
             } finally { 
                scope.fibers().forEach(Fiber::cancel); // cancel remaining fibers
             } 
           } 
        }
        

      • 结构化并发

        • 同样的有截止日期，如果过期则已调度的在作用域的fiber都将被取消

          <V> V anyOf(Callable<? extends V>[] tasks, Instant deadline) throws Throwable { 
              try (var scope = FiberScope.withDeadline(deadline)) { 
                  var queue = new FiberScope.TerminationQueue<V>(); 
                  Arrays.stream(tasks).forEach(task -> scope.schedule(task, queue)); 
                  
              try { 
                  return queue.take().join(); 
               } catch (CompletionException e) { 
                  throw e.getCause(); 
               } finally { 
                  scope.fibers().forEach(Fiber::cancel); // cancel remaining fibers
               } 
             } 
          }
          

  • Generators

    • 我们希望人们使用fiber，而不是直接使用continuation
    • 但有趣的是，continuation的一个用途是类似python的生成器（run,yield,run...）

    class Generator<T> implements Iterable<T> { 
        private static final ContinuationScope GENERATOR = new ContinuationScope(); 
        private static class GenContinuation<T> extends Continuation { 
            public GenContinuation() { super(GENERATOR); } 
            private T value; 
         } 
         
        public static void yield(T value) { 
             ((GenContinuation)currentContinuation(GENERATOR)).val = value; 
            Continuation.yield(GENERATOR); 
        } 
    
        private final GenContinuation<T> cont; 
        public Generator(Runnable body) { 
        cont = new GenContinuation<T>(body);
        } 
        
        public Iterator<T> iterator() { 
            return new Iterator<T>() { 
            public T next() { cont.run(); return cont.val; } 
            public boolean hasNext() { return !cont.isDone(); } 
            } 
       } 
    }
    
    var fibonacci = new Generator<Integer>(() -> { 
        Generator.yield(0); 
        int a = 0; 
        int b = 1; 
        while(true) { 
            Generator.yield(b); 
            var sum = a + b; 
            a = b; 
            b = sum; 
        } 
    });
    
    // 调用迭代器，next和hasNext方法通过continuation的run和yield驱动
    for (var num : fibonacci) { 
        System.out.println(num); 
        if (num > 10_000) break; 
    }
    
    // 另外一个例子
    var greetedPrimes = new Generator<String>(() -> { 
        for (int n = 0; ; n++) 
        if (isPrime(n)) { 
            // 阻塞
            var greeting = Console.readLine(); 
            Generator.yield(greeting + ": " + n); 
        } 
    }); 
    
    Fiber.schedule(()-> { 
        for (var x : greetedPrimes) 
        System.out.println(x); 
    }
    

  • right now mostly interested input on the structure currency API

    • 现在最感兴趣的是结构化并发api的输入