KO | EN

GitLite

Minimalist Open Source Repository Search Engine

gitlite — search

#typescript #ai-agents #cli #claude-code #codex #developer-tools #ai #dsh-plugin #ai-agent #mcp #deepseek-harness #windows

meteor-cluster

★ 18 Open GitHub ↗

worker pool for meteor using node js native `cluster` module

// repository documentation

Was this content helpful?

★ 0 (0 ratings)

# meteor-cluster

Meteor Package enabling users to create a Worker Pool on the server to handle heavy jobs. 
It can run synchronous and asynchronous tasks from a persitent / in-memory queue. 
It can also run recurring and scheduled tasks.

# TaskQueue

`TaskQueue` is both a Mongodb and an in-memory backed job queue. 
 It enables to add, update, remove jobs consistently between processes.

You can attach event listeners to handle the tasks results / errors

## prototype

`TaskQueue.addTask({ taskType: String, data: Object, priority: Integer, _id: String, dueDate: Date, inMemory: Boolean })`
 - `taskType` is mandatory
 - `data` is mandatory but you can pass an empty object
 - `priority` is mandatory, default is set to 1
 - `_id` is optional
 - `dueDate` is mandatory, default is set to `new Date()`
 - `inMemory` is optional, default is set to `false`

### Event listeners (Master only) :

`TaskQueue.addEventListener(eventType: String, callback: function)`
 - `eventType` is one of `[ 'done', 'error' ]`
 - `callback` is a function prototyped as `callback({ value: Any, task: Task })`, `value` contains the result / error.

`TaskQueue.removeEventListener(eventType: String)` 
 - `eventType` is one of `[ 'done', 'error' ]`

note : you can only attach one event listener by eventType.

### In-Memory Queue (Master only) :

`TaskQueue.inMemory.findById(_id: String)` 
 `TaskQueue.inMemory.removeById(_id: String)` 
 `TaskQueue.inMemory.tasks()` : returns all in-memory tasks 
 `TaskQueue.inMemory.availableTasks()` : returns available in-memory tasks

## note on the in-memory / persistent task queue

Both in-memory and persistent tasks are available at the same time, and can be used altogether but :

- in-memory tasks can only be created on the Master (which is because it's non persistent...)
- in-memory tasks will always be called first over persistent tasks even if their respective `priority` are greater.
- if you use both in-memory and persistent tasks at the same time, the persistent tasks will be called only when no in-memory tasks are available (may change later).

# Cluster

`Cluster` is an isomorphic class to handle both the Worker and the Master 
 on the Master it :
 - verifies if jobs are in the queue
 - verifies if workers are available, or create them
 - dispatches jobs to the workers
 - removes the task from the queue once the job is done
 - closes the workers when no jobs are available(behavior can be overriden with `keepAlive`)

on the Worker it :
  - starts the job
  - when the job's done, tell the Master that it's available and to remove previous task.

## prototype

`constructor(taskMap: Object, { port: Integer, maxAvailableWorkers: Integer, refreshRate: Integer, inMemoryOnly: Boolean, messageBroker: function, logs: String, keepAlive: String | Integer, autoInitialize: Boolean })`
 - `taskMap`: a map of functions associated to a `taskType`
 - `maxAvailableWorkers`: maximum number of child process (cores), default is set to system maximum
 - `port`: server port for child process servers, default set to `3008`
 - `refreshRate`: Worker pool refresh rate (in ms), default set to `1000`
 - `inMemoryOnly`: force the cluster to only pull jobs from the in-memory task queue.
 - `messageBroker` is optional, default set to null (see IPC section) 
 - `logs`: is one of `['all', 'error']`, default sets to `all` : if set to `'error'`, will only show the errors and warning logs.
 - `keepAlive`: an optional parameter that can be set to either:
 - `'always'` to have the system start up the `maxAvailableWorkers` number of workers immediately and keep them all alive always
 - some `Integer` value will have the system not shutdown workers until the number passed in milliseconds has passed since last a job was available to be picked up by a worker
 
 **NOTE:** default behavior when `keepAlive` is not set is to only keep alive workers when there are jobs available to be picked up by them. 
 - `autoInitialize`: an optional parameter that controls whether the system will automatically initialize the cluster's polling for jobs when the `MasterCluster` is created. Default is set to `true`.

`Cluster.isMaster()`: `true` if this process is the master

`Cluster.maxWorkers()`: returns the maximum number of workers available at the same time

`setRefreshRate(refreshRate: Integer)`: change the refresh rate on the master

if the Master process crashes or restarts, all the unfinished jobs will be restarted from the beginning. 
 Each job is logged when started / finished with the format : `${timestamp}:nschwarz:cluster:${taskType}:${taskId}`

## IPC (advanced usage)

Introduced in version 2.0.0, you can communicate between the child processes and the Master.
To do so, you must provide the Master Cluster instance with a `messageBroker` function.
this function will handle (on the master) all custom messages from the child processes.

the function should be prototype as follow : 
`messageBroker(respond: function, msg: { status: Int > 1, data: Any })`
- `respond` enables you to answer to a message from a child

All communications between the master and a child must be started by the child.
To do so you can use the second parameter passed in all functions provided to the taskMap `toggleIPC` which is prototyped as follow :

`toggleIPC(messageBroker: function, initalize: function): Promise`
- `messageBroker` is prototyped as `messageBroker(msg: Any)`
- `initialize` is prototyped as `initialize(sendMessageToMaster: function)`

because `toggleIPC` returns a promise you must return it (recursively), otherwise the job will be considered done, and the worker Idle. 
Not returning it will result in unwanted, non expected behavior.

# CPUS allocation

You should not use the default `maxAvailableWorkers` (cpus allocation number) value.
The default value is set to your system cpus number, but it's a reference value.
It's up to you to understand your needs and allocate cpus accordingly.

## how can I calculate the maximum number of cpus I can allocate ?

for example, if you're running on a 8 core machine :

- The app you're running is using 1 cpu to run.
- You should have a reverse proxy on your server, you should at least save 1 cpu (may be more depending on your traffic).
- the database you're using is hosted on the same server, you should save 1 cpu for it.
- you're running an external service such as Redis or Elastic Search, so that's 1 down.

so you should have `maxAvailableWorkers = Cluster.maxWorkers() - 4 === 4`

## what if I allocated too much CPUS ?

You can't allocate more than your maximum system cpu number. 
You still can outrange the theoretical maximum process number :

in such case your overall system should be **slowed down** because some of the processes execution will be deferred.
**It will drastically reduce the multi-core performance gain**.

# examples
## basic usage

```
  import { Meteor } from 'meteor/meteor'
  import { Cluster, TaskQueue } from 'meteor/nschwarz:cluster'

const taskMap = {
    'TEST': job => console.log(`testing ${job._id} at position ${job.data.position}`),
    'SYNC': (job) => console.log("this is a synchronous task"),
    'ASYNC': (job) => new Promise((resolve, reject) => Meteor.setTimeout(() => {
      console.log("this is an asynchronous task")
      resolve()
    }, job.data.timeout))
  }

function onJobsDone({ value, task }) {
    console.log('do something with the result')
  }

function onJobsError({ value, task }) {
    console.log('do something with the error')
  }

function syncTask() {
    return TaskQueue.addTask({ taskType: 'SYNC', data: {}})
  }

function asyncTask() {
    return TaskQueue.addTask({ taskType: 'ASYNC', data: { timeout: 5000 }, priority: 6 })
  }

function inMemoryTask(priority, position) {
   return TaskQueue.addTask({ taskType: 'TEST', priority, data: { position }, inMemory: true })
  }

function persistentTask(priority, position) {
   return TaskQueue.addTask({ taskType: 'TEST', priority, data: { position }, inMemory: false })
  }

const cluster = new Cluster(taskMap)
  Meteor.startup(() => {
    if (Cluster.isMaster()) {
      TaskQueue.addEventListener('done', onJobsDone)
      TaskQueue.addEventListener('error', onJobsError)

syncTask()
      asyncTask()
      inMemoryTask(8, 1)
      inMemoryTask(1, 2)

persistentTask(8, 1)
      persistentTask(1, 2)
    }
  })
```

## scheduled task example : run a task in ten minutes
```
  import { add } from 'date-fns/date' // external library to handle date objects

const dueDate = add(new Date(), { minutes: 10 })
  TaskQueue.addTask({ taskType: 'sometype', priority: 1, data: {}, dueDate })
```

## scheduled task example : run a recurring task every ten minutes
```
  import { add } from 'date-fns/date' // external library to handle date objects

function recurringTask(job) {
    // do something
    const dueDate = add(new Date(), { minutes: 10 })
    TaskQueue.addTask({ taskType: 'recurringTask', priority: 1, data: {}, dueDate })
  }

const taskMap = {
    recurringTask
  }
```

## simple IPC example (advanced usage)
```
function ipcPingTest(job, toggleIPC) {
  return toggleIPC(
    (msg) => {
      console.log(msg)
      return 'result you eventually want to pass to the master'
    }, (smtm) => smtm({ status: 4, data: 'ping' })
  )
}

const taskMap = {
  ipcPingTest
}

function messageBroker(respond, msg) {
  if (msg.data === 'ping') {
    respond('pong')
  }
}

const cluster = new Cluster(taskMap, { messageBroker })
```

## multiple IPC example (advanced usage)
```
function ipcPingTest(job, toggleIPC) {
  return toggleIPC(
    (msg) => {
      console.log(msg)
      return toggleIPC(
        (msg) => console.log(msg),
        (smtm) => smtm({ status: 4, data: 'ping' })
      )
    }, (smtm) => smtm({ status: 4, data: 'ping' }))
}

const taskMap = {
  ipcPingTest
}

function messageBroker(respond, msg) {
  if (msg.data === 'ping') {
    respond('pong')
  }
}

const cluster = new Cluster(taskMap, { messageBroker })
```

# common mistakes and good practices

## secure your imports

Because the worker will only work on tasks, you should remove the unnecessary imports to avoid resources consumption and longer startup time. 
As a good practice you should put all your Master imports logic in the same file, and import it only on the master. 
What I mean by "Master imports Logic" is :

- all your publications
- all your REST endpoints declarations
- graphql server and such...
- SSR / front related code

It could be summarized as such :

```
// in your entry file

if (Cluster.isMaster()) {
  import './MasterImports.js'
}
// ...rest of your cluster logic
```

## recurring tasks

Because recurring tasks are created "recursively", there will always be a task in the queue. 
If the server is restarted, it will start the recurring task because it's still in the queue. 
Be sure to remove all recurring task *on the master* before starting others, or secure the insert. 
Otherwise you will have multiple identical recurring tasks running at the same time.

You can either do :

```
Meteor.startup(() => {
  if (Cluster.isMaster()) {
    TaskQueue.remove({ taskType: 'recurringTask' })
  }  
})
```

or at task *initialization* :

```
  const recurringTaskExists = TaskQueue.findOne({ taskType: 'recurringTask' }) !== undefined
  if (!recurringTaskExists) {
    TaskQueue.addtask({ taskType: 'recurringTask', priority: 1, data: {}, dueDate })
  }
```

## task uniqueness

If you want to be sure to have unique tasks, you should set a unique Id with `TaskQueue.addTask`. 
A good model could be : `${taskType}${associated_Model_ID}`

## multiple apps

There's no way right now to know from which app the task is started (may change later) : 
you should only run the Cluster on **one of the app** to avoid other apps to run a task which is not included in its taskMap. 
You can still use the TaskQueue in all the apps of course. 
If your apps have different domain names / configurations (for the mailer for example), you should pass these through the `data` field.

For example if you're using `Meteor.absoluteUrl` or such in a task it will have the value associated with the app running the Cluster.

meteor-cluster

Explore Similar Repositories

Was this content helpful?

Recent Feedback

Download README