Async access to same object

weaver 0
Viewed 38

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!

race-condition async-await typescript javascript
2 Answers

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.

Mulan0

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);
user32523270
Related