Ali NodeJs framework Egg for node multi-process implementation

Excerpts from: eggjs.org/zh-cn/core/…

We know that JavaScript code runs on a single thread, in other words, a Node.js process can only run on one CPU. So if you use Node.js as a Web Server, you can’t enjoy the benefits of multi-core computing. As an enterprise-level solution, one of the problems we have to solve is:

How to drain server resources and take advantage of concurrency on multi-core cpus?

Node.js officially provides a solution called the Cluster module, which contains a brief introduction:

A single Node.js instance runs in a single-threaded environment. To take advantage of multi-core environments, users sometimes want to start a batch of Node.js processes for loading.

Clustered modules make it easy to create child processes that can be shared between service ports.

What is a Cluster?

To put it simply,

• Start multiple processes simultaneously on the server.
• Each process runs the same source code (like dividing the work of a previous process among multiple processes).
• Amazingly, these processes can listen on the same port at the same time (see @davidcai1993 on Cluster implementation).

Among them:

• The process responsible for starting other processes is called the Master process, it is like a “contractor”, does not do the specific work, only responsible for starting other processes.
• The other processes that are started are called Worker processes, which are literally “workers” doing the work. They receive requests and provide services externally.
• The number of Worker processes is generally determined according to the number of CPU cores on the server, so that multi-core resources can be perfectly utilized.

const cluster = require('cluster'); const http = require('http'); const numCPUs = require('os').cpus().length;if (cluster.isMaster) {  // Fork workers.  for (let i = 0; i < numCPUs; i++) {    cluster.fork();  }  cluster.on('exit'.function(worker, code, signal) {    console.log('worker ' + worker.process.pid + ' died'); }); }else {  // Workers can share any TCP connection  // In this case it is an HTTP server  http.createServer(function(req, res) {    res.writeHead(200);    res.end("hello world\n"); }).listen(8000); }Copy the code

The multi-process model of the framework

The above example isn’t simple, but there’s a lot more to consider as an enterprise-level solution.

• How can I handle the abnormal exit of the Worker process?
• How do multiple Worker processes share resources?
• How to schedule multiple Worker processes?
• .

The process to protect

Robustness (also called robustness) is an issue that enterprise applications must consider. In addition to ensuring the code quality of the program itself, the framework level also needs to provide a corresponding “bottom” mechanism to ensure the availability of the application in extreme cases.

In general, Node.js process exits can be classified into two categories:

Uncaught exception

Node.js provides the process.on(‘uncaughtException’, handler) interface to catch it. However, when a Worker process encounters an uncaughtException, It is already in an indeterminate state, at which point we should gracefully exit the process:

1. Close all TCP servers of the abnormal Worker process (quickly disconnect the existing connection and stop receiving new connections), disconnect the IPC channel with the Master and stop accepting new user requests.
2. The Master immediately forks a new Worker process, keeping the total number of “workers” online unchanged.
3. The abnormal Worker waits for a period of time, processes the accepted requests and exits.

+---------+ +---------+| Worker | | Master |+---------+ +----+----+ | uncaughtException | +------------+ | | | | +---------+ | <----------+ | | Worker | | | +----+----+ | disconnect | fork a new worker | +-------------------------> +  ---------------------> | | wait... | | |exit             |                        |     +-------------------------> |                        |     |                           |                        |    die                          |                        |                                 |                        |                                 |                        |Copy the code

OOM. The system is abnormal

However, when a process is abnormal and crash or OOM is killed by the system, we still have the opportunity to continue the process execution unlike when the exception is not captured. The Master can only make the current process exit directly and fork a new Worker immediately.

In the framework, we use graceful and egg-cluster modules to implement the above logic. This scheme has been widely deployed in the production environment of Alibaba and Ant Financial, and has stood the test of “Double 11” promotion, so it is relatively stable and reliable.

The Agent mechanism

Speaking of which, the Node.js multi-process solution seems to be taking shape, which is what we used online in the early days. However, we later found that some tasks do not need to be done by every Worker. If they are all done, it will waste resources on the one hand and, more importantly, may lead to resource access conflicts between multiple processes. For example: Log files in production are usually archived by date, which is easy to do in a single-process model:

1. At 00:00 every day, the current log file is renamed according to the date
2. Destroy the previous file handle and create a new log file to continue writing

Imagine if there were now four processes doing the same thing. Therefore, for this kind of background running logic, we hope to put them into a separate process to execute, which is called Agent Worker, Agent for short. Agent is like a “secretary” hired by the Master for other workers. It does not provide external services but only works for App workers, specializing in handling some public affairs. Now our multi-process model looks like this

                +--------+          +-------+                | Master |<-------->| Agent |                +--------+          +-------+                ^   ^    ^               /    |     \             /      |       \           /        |         \         v          v          v+----------+   +----------+   +----------+| Worker 1 |   | Worker 2 |   | Worker 3 |+----------+   +----------+   +----------+Copy the code

The startup sequence of our framework is as follows:

+---------+ +---------+ +---------+| Master | | Agent | | Worker |+---------+ +----+----+ +----+----+ | fork agent | | +-------------------->| | | agent ready | | |<--------------------+ | | | fork worker | +----------------------------------------->| | worker ready | | |<-----------------------------------------+ | Egg ready  | | +-------------------->| | | Egg ready | | +----------------------------------------->|Copy the code

1. After the Master starts, fork the Agent process
2. After the Agent is successfully initialized, the Master is notified through the IPC channel
3. The Master forks multiple App workers
4. Notify the Master that the App Worker is successfully initialized
5. After all processes are successfully initialized, the Master notifies the Agent and Worker that the application is successfully started

In addition, there are several points to note about Agent workers:

1. Since App Worker depends on Agent, fork App Worker only after Agent initialization is complete
2. Although the Agent is the “secret” of App Worker, it should not be placed on the Agent to do business-related work, otherwise it will be bad for her to be exhausted
3. Due to the special positioning of Agent, we should ensure its relative stability. When an uncaught exception occurs, the framework will not let it exit and restart like App Worker, but record the exception log, alarm and wait for manual processing
4. The API mounted on Agent and common App Worker is not exactly the same. How to identify the difference can be found in the framework document

The use of the Agent

You can implement your own logic in agent.js at the root of your application or plug-in (similar to launching custom, except that the entry argument is an Agent object).

Jsmodule. Exports = agent => {// write your initialization logic here // You can also send messages to the App Worker via messenger object // wait for the App Worker Agent. Messenger. On ('egg-ready', () => {    const data = { ... };    agent.messenger.sendToApp('xxx_action', data); }); };Copy the code

// app.jsmodule.exports = app => {  app.messenger.on('xxx_action', data => { // ... }); };Copy the code

In this example, the agent.js code will be executed on the Agent process, and the app.js code will be executed on the Worker process. They conduct inter-process communication (IPC) through the messenger object encapsulated by the framework. The following chapters will explain the IPC of the framework in detail.

Master VS Agent VS Worker

When an application is started, all three processes are started simultaneously.

Master

In this model, the Master process undertakes the work of process management (similar to PM2), without running any business code, we only need to run a Master process, which will help us handle the initialization and restart of all Worker and Agent processes.

The stability of the Master process is extremely high. When running online, we only need to run the Master process started by the egg.startCluster through the egg-scripts background. There is no need to use pm2 and other daemon modules.

$ egg-scripts start --daemonCopy the code

Agent

In most cases, the Agent process is completely ignored when writing business code, but it comes into play when there are scenarios where you just want the code to run on one process.

Because the Agent has only one Agent and is responsible for many dirty and tiring tasks to maintain connections, it cannot be easily hung up or restarted. Therefore, the Agent process will not exit when listening to uncaught exceptions, but will print error logs. We need to be vigilant about uncaught exceptions in logs.

Worker

The Worker process handles real user requests and the processing of scheduled tasks. The scheduled task of Egg also provides the ability to run only one Worker process, so the problems that can be solved by the scheduled task should not be executed on the Agent.

Worker runs business codes, which are relatively more complex and less stable than the codes running on Agent and Master processes. When the Worker process abnormally exits, the Master process will restart a Worker process.

Interprocess Communication (IPC)

Although each Worker process is relatively independent, they still need to communicate with each other, which is called inter-process communication (IPC). Here is a sample code provided by Node.js

'use strict'; const cluster = require('cluster');if (cluster.isMaster) {  const worker = cluster.fork();  worker.send('hi there');  worker.on('message', msg => {    console.log(`msg: ${msg} from worker#${worker.id}`); }); } else if (cluster.isWorker) { process.on('message', (msg) => { process.send(msg); }); }Copy the code

If you are careful, you may have found that the IPC channel of cluster only exists between Master and Worker/Agent, and Worker and Agent processes do not exist between each other. So what should workers do to communicate with each other? Yes, through Master.

Broadcast message:  agent => all workers +--------+ +-------+ | Master |<---------| Agent | +--------+ +-------+ / | \ / | \ / | \ / | \ v v v +----------+ +----------+ +----------+ | Worker 1 | | Worker 2 | | Worker 3 | +----------+ +----------+ +----------+ Designated receiver:  one worker => another worker +--------+ +-------+ | Master |----------| Agent | +--------+ +-------+ ^ | send to / | worker 2 / | / | / v +----------+ +----------+ +----------+ | Worker 1 | | Worker 2 | | Worker 3 | +----------+ +----------+ +----------+Copy the code

To make it easier to call, we’ve wrapped a Messenger object to hang on app/Agent instances, providing a set of friendly apis.

send

• app.messenger.broadcast(action, data): Send to all Agent/app processes (including yourself)
• app.messenger.sendToApp(action, data): sent to all app processes
  • Calling this method on your app sends it to yourself and other app processes
  • Calling this method on the Agent is sent to all app processes
• app.messenger.sendToAgent(action, data): Sent to the Agent process
  • Calling this method on the APP sends it to the Agent process
  • Calling this method on the Agent sends it to the agent itself
• agent.messenger.sendRandom(action, data):
  • This method is not available in the app (currently Egg implementation is equivalent to sentToAgent)
  • The Agent sends a random message to an app process (the master controls who the message is sent to)
• app.messenger.sendTo(pid, action, data): Sent to the specified process

// app.jsmodule.exports = app => {// Note that the message app.messenger. Once (// app.jsmodule. Exports = app => {// note that the message can only be sent after the egg-ready event is received.'egg-ready', () => {    app.messenger.sendToAgent('agent-event', { foo: 'bar' });    app.messenger.sendToApp('app-event', { foo: 'bar'}); }); }Copy the code

egg-ready

As mentioned in the example above, you need to wait for the egg-ready message before sending it. Only after the Master confirms that all Agent and Worker processes are successfully started (and ready) will the Master send an egg-ready message to all agents and workers via Messenger to inform them that everything is ready. The IPC channel is ready to be used.

receive

Listen for action events on Messenger to receive messages from other processes.

app.messenger.on(action, data => { // process data}); app.messenger.once(action, data => { // process data});Copy the code

The IPC of actual combat

Let’s use a simple example to get a feel for how IPC can be used to solve a real-world problem under the framework’s multi-process model.

demand

We have an interface that needs to read some data from a remote data source, providing apis to the outside world, but the data from this data source rarely changes, so we want to cache the data in memory to improve service capacity and reduce RT. There needs to be a mechanism for updating the memory cache.

1. Periodically obtain data from the remote data source and update the memory cache. To reduce the pressure on the data source, the update interval is set to be long.
2. The remote data source provides an interface to check for data updates, and our service can call the interface more frequently to pull data when it is updated.
3. Remote data sources push messages of data updates through the messaging middleware, and our service listens for messages to update the data.

In the actual project, we can adopt plan 1 for the bottom, and combine plan 3 or Plan 2 to improve the real-time performance of data update. In this example, we implement all three cache update schemes simultaneously through IPC + scheduled tasks.

implementation

We encapsulate all the logic that interacts with the remote data source in a Service and provide the GET method for the Controller to call.

// app/service/source.jslet memoryCache = {}; class SourceService extends Service { get(key) {return memoryCache[key];  }  async checkUpdate() {    // check if remote data source has changed    const updated = await mockCheck();    this.ctx.logger.info('check update response %s', updated);    return updated;  }  async update() {    // update memory cache from remote    memoryCache = await mockFetch();    this.ctx.logger.info('update memory cache from remote: %j', memoryCache);  }}Copy the code

Write a scheduled task to achieve scheme 1, which periodically obtains data from the remote data source every 10 minutes to update the cache as the bottom.

// app/schedule/force_refresh.jsexports.schedule = {  interval: '10m'.type: 'all', // run inall workers}; exports.task = async ctx => { await ctx.service.source.update(); ctx.app.lastUpdateBy ='force'; };Copy the code

Another scheduled task is written to realize the check logic of scheme 2. Let a worker call the check interface every 10s. When data changes are found, all workers are notified through the method provided by Messenger.

// app/schedule/pull_refresh.jsexports.schedule = {  interval: '10s'.type: 'worker', // only run inone worker}; exports.task = async ctx => { const needRefresh = await ctx.service.source.checkUpdate();if(! needRefresh)return;  // notify all workers to update memory cache from `file`  ctx.app.messenger.sendToApp('refresh'.'pull'); };Copy the code

Listen for refresh event in the startup custom file and update data. All Worker processes can receive the message and trigger the update. At this point, scheme 2 has been completed.

// app.jsmodule.exports = app => {  app.messenger.on('refresh', by => {    app.logger.info('start update by %s', by); // create an anonymous context to access service const ctx = app.createAnonymousContext(); ctx.runInBackground(async () => { await ctx.service.source.update(); app.lastUpdateBy = by; }); }); };Copy the code

Now let’s look at how to implement the third scenario. We need to have a client of message-oriented middleware, which will maintain a long connection with the server. This kind of long connection maintenance is more suitable for the Agent process, which can effectively reduce the number of connections and reduce the consumption of both ends. So we enable message listening on the Agent process.

// agent.jsconst Subscriber = require('./lib/subscriber'); module.exports = agent => { const subscriber = new Subscriber(); // listen changed event, broadcast to all workers subscriber.on('changed', () => agent.messenger.sendToApp('refresh'.'push')); };Copy the code

With the proper use of Agent processes, scheduled tasks, and IPC, we can easily address similar requirements and reduce pressure on data sources. See examples/ipc for examples of code.