深刻理解ES8的新特性SharedArrayBuffer

简介

ES8引入了SharedArrayBuffer和Atomics，经过共享内存来提高workers之间或者worker和主线程之间的消息传递速度。

本文将会详细的讲解SharedArrayBuffer和Atomics的实际应用。

Worker和Shared memory

在nodejs中，引入了worker_threads模块，能够建立Worker. 而在浏览器端，能够经过web workers来使用Worker(）来建立新的worker。

这里咱们主要关注一下浏览器端web worker的使用。

咱们看一个常见的worker和主线程通讯的例子，主线程：

var w = new Worker("myworker.js")

w.postMessage("hi");     // send "hi" to the worker
w.onmessage = function (ev) {
  console.log(ev.data);  // prints "ho"
}

myworker的代码：

onmessage = function (ev) {
  console.log(ev.data);  // prints "hi"
  postMessage("ho");     // sends "ho" back to the creator
}

咱们经过postMessage来发送消息，经过onmessage来监听消息。

消息是拷贝以后，通过序列化以后进行传输的。在解析的时候又会进行反序列化，从而下降了消息传输的效率。

为了解决这个问题，引入了shared memory的概念。

咱们能够经过SharedArrayBuffer来建立Shared memory。

考虑下上面的例子，咱们可把消息用SharedArrayBuffer封装起来，从而达到内存共享的目的。

//发送消息
var sab = new SharedArrayBuffer(1024);  // 1KiB shared memory
w.postMessage(sab)

//接收消息
var sab;
onmessage = function (ev) {
   sab = ev.data;  // 1KiB shared memory, the same memory as in the parent
}

上面的这个例子中，消息并无进行序列化或者转换，都使用的是共享内存。

ArrayBuffer和Typed Array

SharedArrayBuffer和ArrayBuffer同样是最底层的实现。为了方便程序员的使用，在SharedArrayBuffer和ArrayBuffer之上，提供了一些特定类型的Array。好比Int8Array，Int32Array等等。

这些Typed Array被称为views。

咱们看一个实际的例子，若是咱们想在主线程中建立10w个质数，而后在worker中获取这些质数该怎么作呢？

首先看下主线程：

var sab = new SharedArrayBuffer(Int32Array.BYTES_PER_ELEMENT * 100000); // 100000 primes
var ia = new Int32Array(sab);  // ia.length == 100000
var primes = new PrimeGenerator();
for ( let i=0 ; i < ia.length ; i++ )
   ia[i] = primes.next();
w.postMessage(ia);

主线程中，咱们使用了Int32Array封装了SharedArrayBuffer，而后用PrimeGenerator来生成prime，存储到Int32Array中。

下面是worker的接收：

var ia;
onmessage = function (ev) {
  ia = ev.data;        // ia.length == 100000
  console.log(ia[37]); // prints 163, the 38th prime
}

并发的问题和Atomics

上面咱们获取到了ia[37]的值。由于是共享的，因此任何可以访问到ia[37]的线程对该值的改变，均可能影响其余线程的读取操做。

好比咱们给ia[37]从新赋值为123。虽然这个操做发生了，可是其余线程何时可以读取到这个数据是未知的，依赖于CPU的调度等等外部因素。

为了解决这个问题，ES8引入了Atomics,咱们能够经过Atomics的store和load功能来修改和监控数据的变化：

console.log(ia[37]);  // Prints 163, the 38th prime
Atomics.store(ia, 37, 123);

咱们经过store方法来向Array中写入新的数据。

而后经过load来监听数据的变化：

while (Atomics.load(ia, 37) == 163)
  ;
console.log(ia[37]);  // Prints 123

还记得java中的重排序吗？

在java中，虚拟机在不影响程序执行结果的状况下，会对java代码进行优化，甚至是重排序。最终致使在多线程并发环境中可能会出现问题。

在JS中也是同样，好比咱们给ia分别赋值以下：

ia[42] = 314159;  // was 191
ia[37] = 123456;  // was 163

按照程序的书写顺序，是先给42赋值，而后给37赋值。

console.log(ia[37]);
console.log(ia[42]);

可是由于重排序的缘由，可能37的值变成123456以后，42的值仍是原来的191。

咱们可使用Atomics来解决这个问题，全部在Atomics.store以前的写操做，在Atomics.load发送变化以前都会发生。也就是说经过使用Atomics能够禁止重排序。

ia[42] = 314159;  // was 191
Atomics.store(ia, 37, 123456);  // was 163

while (Atomics.load(ia, 37) == 163)
  ;
console.log(ia[37]);  // Will print 123456
console.log(ia[42]);  // Will print 314159

咱们经过监测37的变化，若是发生了变化，则咱们能够保证以前的42的修改已经发生。

一样的，咱们知道在java中++操做并非一个原子性操做，在JS中也同样。

在多线程环境中，咱们须要使用Atomics的add方法来替代++操做，从而保证原子性。

注意，Atomics只适用于Int8Array, Uint8Array, Int16Array, Uint16Array, Int32Array or Uint32Array。

上面例子中，咱们使用while循环来等待一个值的变化，虽然很简单，可是并非颇有效。

while循环会占用CPU资源，形成没必要要的浪费。

为了解决这个问题，Atomics引入了wait和wake操做。

咱们看一个应用：

console.log(ia[37]);  // Prints 163
Atomics.store(ia, 37, 123456);
Atomics.wake(ia, 37, 1);

咱们但愿37的值变化以后通知监听在37上的一个数组。

Atomics.wait(ia, 37, 163);
console.log(ia[37]);  // Prints 123456

当ia37的值是163的时候，线程等待在ia37上。直到被唤醒。

这就是一个典型的wait和notify的操做。

使用Atomics来建立lock

咱们来使用SharedArrayBuffer和Atomics建立lock。

咱们须要使用的是Atomics的CAS操做：

compareExchange(typedArray: Int8Array | Uint8Array | Int16Array | Uint16Array | Int32Array | Uint32Array, index: number, expectedValue: number, replacementValue: number): number;

只有当typedArray[index]的值 = expectedValue 的时候，才会使用replacementValue来替换。 同时返回typedArray[index]的原值。

咱们看下lock怎么实现：

const UNLOCKED = 0;
const LOCKED_NO_WAITERS = 1;
const LOCKED_POSSIBLE_WAITERS = 2;

    lock() {
        const iab = this.iab;
        const stateIdx = this.ibase;
        var c;
        if ((c = Atomics.compareExchange(iab, stateIdx,
        UNLOCKED, LOCKED_NO_WAITERS)) !== UNLOCKED) {
            do {
                if (c === LOCKED_POSSIBLE_WAITERS
                || Atomics.compareExchange(iab, stateIdx,
                LOCKED_NO_WAITERS, LOCKED_POSSIBLE_WAITERS) !== UNLOCKED) {
                    Atomics.wait(iab, stateIdx,
                        LOCKED_POSSIBLE_WAITERS, Number.POSITIVE_INFINITY);
                }
            } while ((c = Atomics.compareExchange(iab, stateIdx,
            UNLOCKED, LOCKED_POSSIBLE_WAITERS)) !== UNLOCKED);
        }
    }

UNLOCKED表示目前没有上锁，LOCKED_NO_WAITERS表示已经上锁了，LOCKED_POSSIBLE_WAITERS表示上锁了，而且还有其余的worker在等待这个锁。

iab表示要上锁的SharedArrayBuffer，stateIdx是Array的index。

再看下tryLock和unlock:

tryLock() {
        const iab = this.iab;
        const stateIdx = this.ibase;
        return Atomics.compareExchange(iab, stateIdx, UNLOCKED, LOCKED_NO_WAITERS) === UNLOCKED;
    }

    unlock() {
        const iab = this.iab;
        const stateIdx = this.ibase;
        var v0 = Atomics.sub(iab, stateIdx, 1);
        // Wake up a waiter if there are any
        if (v0 !== LOCKED_NO_WAITERS) {
            Atomics.store(iab, stateIdx, UNLOCKED);
            Atomics.wake(iab, stateIdx, 1);
        }
    }

使用CAS咱们实现了JS版本的lock。

固然，有了CAS，咱们能够实现更加复杂的锁操做，感兴趣的朋友，能够自行探索。

本文做者：flydean程序那些事

本文连接：http://www.flydean.com/es8-shared-memory/

本文来源：flydean的博客

欢迎关注个人公众号:「程序那些事」最通俗的解读，最深入的干货，最简洁的教程，众多你不知道的小技巧等你来发现！