Async access to same object

Question

I've been using JS/TS for a while but the existence or not of race conditions is stil confusing me. I am trying to implement the following logic:

class Deployer {
 protected txDataList: Array<TXData>;
 
 constructor() {
   this.txDataList = new Array<TXData>();
 }
 async consumeTXs() {
   if (this.txDataList && this.txDataList.length > 0) {
     // dom something
   }
}
setup() {
   // listens for txData events emitted by a service 
  // and pushes them to the txDataList
  this.service.on("txData", async (txData) => {
      this.txDataList.push(txData);
 });
}
start() {
  setup();
  setInterval(this.consumeTXs, 5000);
 }
}

In short, we need to listen for events emitted by some service, put them in an array and every 5 secs we must take whatever events have been pushed in the array and process them. We I run this, the consumeTXs functions runs indeed every 5s but it always "sees" an emtpy list, whereas the setup function does indeed pick up the txData events and puts them in the array. It is as if the two functions see a different version of the txDataList object.

Obviously the implementation of what I described above is wrong. This code would actually lead to a race condition. What is the correct way to implement what I describe and also why does the above code not work? Currious to know what happends under the hood.

Thank you!

your first issue is losing `this` context on `setInterval(this.consumeTXs,...)`, try `setInterval(_ => this.consumeTXs(),...)` or `setInterval(this.consumeTXs.bind(this),...)` — Mulan, Sep 20 '22 at 00:21
"*This code would actually lead to a race condition*" - what makes you think so? Unfortunately you have not shared what "*`// dom something`*" really does. — Bergi, Sep 20 '22 at 00:36
I agree wtih @Bergi. If you are doing `Array.push()` to populate the list and you are doing `Array.shift()` to take things off the list, I don't see any cause for concern. I would be concerned if you are rewriting the list. There's not enough code to make that assessment. — Stephen Quan, Sep 20 '22 at 00:40
I've removed the `setInterval` problem from your code, since it distracts from the question about race conditions. If this was your only problem, the question would be a duplicate of https://stackoverflow.com/q/2749244/1048572, https://stackoverflow.com/q/7890685/1048572, https://stackoverflow.com/q/20279484/1048572, and many more — Bergi, Sep 20 '22 at 00:47
Actually, I am just reading the list in the missing sagment. You are right, it would not technically cause a race condition. The issue as the lost this context. — weaver, Sep 20 '22 at 00:51

Mulan · Answer 1 · 2022-09-20T01:32:09.407

lock/unlock

You could try adding a "lock" state to the Deployer -

// Deployer.js

class Deployer {
  protected txDataList: Array<TXData>
  protected isLocked: Boolean           // lock state
  constructor() {
    this.txDataList = []
    this.isLocked = false
    setInterval(_ => this.read(), 5000)
  }
  async read() {
    if (this.isLocked) return     // if locked, exit
    this.isLocked = true          // lock
    let tx
    while (true) {
      tx = this.txDataList.shift()// fifo tx
      if (tx == null) break       // if no tx, stop processing
      await doSomething(tx)       // process each tx
    }
    this.isLocked = false         // unlock
  }
  write(tx: TXData) {
    this.txDataList.push(tx)
  }
}

export default Deployer

// some-service.js

import Deployer from "./Deployer.js"
const deployer = new Deployer()
someService.on("txData", tx => deployer.write(tx))

So we can write to the deployer whenever, but read only functions when the deployer is in an "unlocked" state.

buffer

Maybe Buffer is a better name because it describes the behavior of the class. TXData can also be abstracted so that we could use our Buffer for any data. And let's make interval a parameter -

// Buffer.js

class Buffer<T> { // generic T
  private buffer: Array<T> // generic T
  private isLocked: Boolean
  constructor({ interval = 5 }) { // make interval a parameter
    this.buffer = []
    this.isLocked = false
    setInterval(_ => this.read(), interval * 1000)
  }
  async read() {
    if (this.isLocked) return
    this.isLocked = true
    let t
    while (true) {
      t = this.buffer.shift()
      if (t == null) break
      await doSomething(t)
    }
    this.isLocked = false
  }
  write(t: T) {  // generic T
    this.buffer.push(t)
  }
}

Now specify the type for Buffer<T> as Buffer<TXData>. Optionally set an interval when constructing the Buffer -

// some-service.js

import Buffer from "./Buffer.js"
const buffer = new Buffer<TXData>({ iterval: 10 }) 
someService.on("txData", tx => buffer.write(tx))

move effect to call site

Next we should get rid of that hard-coded doSomething. Make onRead a parameter so the caller can decide to do with each item -

// Buffer.js

class Buffer<T> { 
  private buffer: Array<T> 
  private isLocked: Boolean
  private onRead: (x: T) => void                        // onRead
  constructor({ interval = 5, onRead = console.log }) { // onRead
    this.buffer = []
    this.isLocked = false
    setInterval(_ => this.read(), interval * 1000)
  }
  async read() {
    if (this.isLocked) return
    this.isLocked = true
    let t
    while (true) {
      t = this.buffer.shift()
      if (t == null) break
      await this.onRead(t)                               // onRead
    }
    this.isLocked = false
  }
  write(t: T) {
    this.buffer.push(t)
  }
}

Caller decides the doSomething effect -

// some-service.js

import Buffer from "./Buffer.js"

async function doSomething(tx: TXData) {
  await // process tx ...
}

const buffer = new Buffer<TXData>({
  interval: 10,
  onRead: doSomething    // specify onRead
})

someService.on("txData", tx => buffer.write(tx))

related

See this Q&A for a related channel abstraction using async generators.

score 0 · Answer 2 · answered Sep 20 '22 at 00:40

Your issue here comes from setInterval(this.consumeTXs, 5000);. By doing so when you run consumeTXs the this context is not your class instance but the window. A fix to this could be to use an arrow function to preserve your instance context:

setInterval(() => this.consumeTXs(), 5000);

Async access to same object

2 Answers2