Unverified Commit f4e245b4 authored by Hong Minhee's avatar Hong Minhee
Browse files

Parallelize task workers 

Close https://github.com/dahlia/fedify/issues/106
parent 276c412a

.vscode/settings.json

+1 −0
Original line number Diff line number Diff line
@@ -127,6 +127,7 @@ 
    "superproperty",

    "supertypes",

    "tempserver",

    "ts-nocheck",

    "twoslash",

    "typeof",

    "unfollow",

CHANGES.md

+5 −0
Original line number Diff line number Diff line
@@ -109,10 +109,14 @@ To be released. 
    the activity to be sent only once.  It had added Object Integrity Proofs

    to the activity for every recipient before.



 -  Added `ParallelMessageQueue` class.  [[#106]]



 -  WebFinger responses now include <http://webfinger.net/rel/avatar> links

    if the `Actor` object returned by the actor dispatcher has `icon`/`icons`

    property.



 -  `DenoKvMessageQueue` now implements `Disposable` interface.



 -  The `fedify inbox` command now sends `Delete(Application)` activities when

    it's terminated so that the peers can clean up data related to the temporary

    actor.  [[#135]]
@@ -123,6 +127,7 @@ To be released. 
     -  `["fedify", "sig", "ld"]`



[Linked Data Signatures]: https://web.archive.org/web/20170923124140/https://w3c-dvcg.github.io/ld-signatures/

[#106]: https://github.com/dahlia/fedify/issues/106

[#135]: https://github.com/dahlia/fedify/issues/135

[#137]: https://github.com/dahlia/fedify/issues/137

docs/manual/inbox.md

+10 −0
Original line number Diff line number Diff line
@@ -241,6 +241,16 @@ the `createFederation()` function. 
> [!NOTE]

> Activities with invalid signatures/proofs are silently ignored and not queued.



> [!TIP]

> If your inbox listeners are mostly I/O-bound, consider parallelizing

> message processing by using the `ParallelMessageQueue` class.  For more

> information, see the [*Parallel message processing*

> section](./mq.md#parallel-message-processing).

>

> If your inbox listeners are CPU-bound, consider running multiple nodes of

> your application so that each node can process messages in parallel with

> the shared message queue.



[`RedisMessageQueue`]: https://jsr.io/@fedify/redis/doc/mq/~/RedisMessageQueue

[`@fedify/redis`]: https://github.com/dahlia/fedify-redis

docs/manual/mq.md

+46 −3
Original line number Diff line number Diff line
@@ -52,7 +52,7 @@ const federation = createFederation<void>({ 
// ---cut-start---

  kv: null as unknown as KvStore,

// ---cut-end---

  queue: new InProcessMessageQueue(),

  queue: new InProcessMessageQueue(),  // [!code highlight]

  // ... other options

});

~~~~
@@ -79,7 +79,7 @@ import { DenoKvMessageQueue } from "@fedify/fedify/x/deno"; 


const kv = await Deno.openKv();

const federation = createFederation<void>({

  queue: new DenoKvMessageQueue(kv),

  queue: new DenoKvMessageQueue(kv),  // [!code highlight]

  // ... other options

});

~~~~
@@ -116,7 +116,7 @@ const federation = createFederation<void>({ 
// ---cut-start---

  kv: null as unknown as KvStore,

// ---cut-end---

  queue: new RedisMessageQueue(() => new Redis()),

  queue: new RedisMessageQueue(() => new Redis()),  // [!code highlight]

  // ... other options

});

~~~~
@@ -179,3 +179,46 @@ custom `MessageQueue`: 


However, you don't need to implement retry logic yourself, as Fedify handles

retrying failed messages automatically.





Parallel message processing

---------------------------



*This API is available since Fedify 1.0.0.*



Fedify supports parallel message processing by running multiple workers

concurrently.  To enable parallel processing, wrap your `MessageQueue` with

`ParallelMessageQueue`, a special implementation of the `MessageQueue` interface

designed to process messages in parallel.  It acts as a decorator for another

`MessageQueue` implementation, allowing for concurrent processing of messages

up to a specified number of workers:



~~~~ typescript twoslash

import type { KvStore } from "@fedify/fedify";

// ---cut-before---

import { createFederation, ParallelMessageQueue } from "@fedify/fedify";

import { RedisMessageQueue } from "@fedify/redis";

import Redis from "ioredis";



const baseQueue = new RedisMessageQueue(() => new Redis());



// Use parallelQueue in your Federation configuration

const federation = createFederation<void>({

  queue: new ParallelMessageQueue(baseQueue, 5),  // [!code highlight]

  // ... other options

  // ---cut-start---

  kv: null as unknown as KvStore,

  // ---cut-end---

});

~~~~



> [!NOTE]

> The workers do not run in truly parallel, in the sense that they are not

> running in separate threads or processes.  They are running in the same

> process, but are scheduled to run in parallel.  Hence, this is useful for

> I/O-bound tasks, but not for CPU-bound tasks, which is okay for Fedify's

> workloads.

>

> If your [inbox listeners](./inbox.md) are CPU-bound, you should consider

> running multiple nodes of your application so that each node can process

> messages in parallel with the shared message queue.

docs/package.json

+1 −1
Original line number Diff line number Diff line 
{

  "devDependencies": {

    "@deno/kv": "^0.8.2",

    "@fedify/fedify": "^1.0.0-dev.404",

    "@fedify/fedify": "1.0.0-dev.407",

    "@fedify/redis": "^0.1.1",

    "@hono/node-server": "^1.12.2",

    "@js-temporal/polyfill": "^0.4.4",