Concurrent async producers, an experiment with ImmerJS

If there's one thing Redux-like libraries don't reduce, it's the amount of boilerplate code you need to write before your app does anything useful. And most developers don't like writing boilerplate as evidenced by the surge in popularity of ImmerJS, which automates the process of writing reducers.

What if we could go one step further, and automate the effects + actions + reducer paradigm ?

The problem with async producers

Unfortunately, the following doesn't work with Immer:

1
2
3
4
5
// Fetch user
newState = await produce(currentState, async draft => {
  const user = await fetchUser();
  draft.user = user;
})
1
2
3
4
5
// Fetch articles
newState = await produce(currentState, async draft => {
  const articles = await fetchArticles();
  draft.articles = articles;
})

The output state will either be

1
2
3
4
5
{ user: { id: '1', name: 'John' } }

**or**

{ articles: [{ id: '1', title: 'First article' }, { id: '1', title: 'Second article' }] }

These async reducers will end up overwriting each other if they are run concurrently. The new state will contain either user or articles but not both, depending on whichever producer completes last. Because of this, it's usually not a good idea to write async producers.

Concurrent producers

With concurrent producers, async actions share the same draft when they run concurrently.

const concurrentProduce = concurrentProduceFactory(initialState);

newState = await concurrentProduce(async draft => {
  const user = await fetchUser();
  draft.user = user;
})

newState = await concurrentProduce(async draft => {
  const articles = await fetchArticles();
  draft.articles = articles;
})

Output state will correctly be reduced to:

1
2
3
4
{
    user: { id: '1', name: 'John' },
    articles: [{ id: '1', title: 'First article' }, { id: '1', title: 'Second article' }],
}

These producers are comparable to composing actions, effects, and pure reducers with libraries like Thunk / Saga / NGRX, all in one (impure) action.

  • Synchronous portions of the async producer represent your pure reducers
  • They can be interleaved with side effects e.g. http calls
  • State updates are batched, the latest state is emitted once an async producer terminates
    • This means updates done in a synchronous section might be sent before its scope completes.
    • They should leave the store in a consistent state (just like you'd expect of any reducer).

Concurrent producer factory

Stackblitz: https://stackblitz.com/edit/immer-concurrent-producer-last-hvh3xn?file=index.ts

import { produce, current, createDraft, finishDraft } from "immer";
import { Draft } from "immer/dist/internal";

// Allow concurrent async actions to share the same draft
// Draft is finalized only once the task queue is empty
function concurrentProduceFactory<T>(initialState: T) {
  let sharedDraft: Draft<T> = undefined;
  let pendingTasks = 0;
  let state: T = initialState;

  return async (recipe: (draft: Draft<T>) => Promise<void>) => {
    pendingTasks++;
    if (!sharedDraft) {
      sharedDraft = createDraft(state);
    }
    try {
      await recipe(sharedDraft);
    } finally {
      pendingTasks--;
      if (pendingTasks === 0) {
        // Finalize the draft only when all tasks are done
        const result = finishDraft(sharedDraft);
        sharedDraft = null;
        return result;
      } else {
        // Until then, emit a snapshot and keep accumulating
        // changes in the current draft
        return current(sharedDraft);
      }
    }
  };
}

/* Usage */

const initialState = {
  property1: 0,
  property2: 0,
  nestedProperty: {
    property1: 0
  }
};

const concurrentProduce = concurrentProduceFactory(initialState);

concurrentProduce(async draft => {
  return new Promise(resolve => {
    setTimeout(() => {
      draft.property1 = 1;
      draft.property2 = 2;
      resolve();
    }, 100);
  });
}).then(state => {
  console.log("Concurrent produce #1");
  console.log(state);
});

concurrentProduce(async draft => {
  const nestedProperty = draft.nestedProperty;
  return new Promise(resolve => {
    setTimeout(() => {
      // Still a valid reference
      nestedProperty.property1 = 1;
      draft.property1 = 3;
      // draft.property2 is 2
      resolve();
    }, 200);
  });
}).then(state => {
  console.log("Concurrent produce #2");
  console.log(state);
});

produce(initialState, async draft => {
  return new Promise(resolve => {
    setTimeout(() => {
      draft.property1 = 1;
      // draft.property2 is 0
      resolve();
    }, 300);
  });
}).then(state => {
  console.log("Async produce #1");
  console.log(state);
});

produce(initialState, async draft => {
  return new Promise(resolve => {
    setTimeout(() => {
      draft.property2 = 2;
      // draft.property1 is 0
      resolve();
    }, 400);
  });
}).then(state => {
  console.log("Async produce #2");
  console.log(state);
});

Considerations

Limitations include:

  • State updates must be done through the Immer proxy. It's not possible to return an entirely new state as there might be other pending tasks relying on the current draft.
  • This is a performance trade-off, since only the last async producer in the current stack will finalize the draft. Other actions will emit the current state represented in the draft, which may be a less efficient operation than finalizing the draft.
  • Draft will accumulate changes until the queue is empty, so it is not a good solution if async actions are running continuously
  • Objects produced through current may not be freezed.
  • It's not clear how safe it is to share the draft, potential errors could invalidate all current async actions

The overhead of always going through Immer proxy and using current() is mitigated by the fact that state updates are now batched. Concurrent actions will only produce the new state once they complete. With Thunk / Saga / NGRX etc. the store is continuously updated which can lead to unnecessary re-render.

Going forward

Ideally you would want a function to finalize the current draft without revoking the proxy. This would essentially create a commit point and restart the draft from the current state, instead of accumulating the mutations in the same draft until all tasks complete.
This could be implemented using a concurrent proxy wrapper, but perhaps a lower level Immer API is preferable, something like state = commitDraft(draf).

bd2f6a3894

Edit

Pub: 16 Mar 2021 19:58 UTC

Edit: 17 Mar 2021 00:29 UTC

Views: 900