写点什么

深入理解 ES8 的新特性 SharedArrayBuffer

发布于: 2021 年 03 月 23 日

简介

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 workerw.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 memoryw.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 primesvar ia = new Int32Array(sab);  // ia.length == 100000var 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 primeAtomics.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 191ia[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 191Atomics.store(ia, 37, 123456);  // was 163
while (Atomics.load(ia, 37) == 163) ;console.log(ia[37]); // Will print 123456console.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 163Atomics.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 的博客

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


发布于: 2021 年 03 月 23 日阅读数: 7
用户头像

关注公众号:程序那些事,更多精彩等着你! 2020.06.07 加入

最通俗的解读,最深刻的干货,最简洁的教程,众多你不知道的小技巧,尽在公众号:程序那些事!

评论

发布
暂无评论
深入理解ES8的新特性SharedArrayBuffer