This project is an implementation of Inter-Process Communication (IPC) in Python using the multiprocessing module, where each worker process runs in parallel and communicates through Queues.
This structure is suitable for systems such as:
- WhatsApp Client Instance Manager
- Task Dispatcher (e.g., sending messages, fetching status, etc.)
- Background Worker Engine
IPC is a mechanism that allows independently running processes to communicate and exchange data.
In multiprocessing:
- Each process has its own memory space.
- Variables between processes cannot be accessed directly.
- A communication channel is needed (such as socket, pipe, or queue).
| Method | Description | Best For |
|---|---|---|
multiprocessing.Queue |
Thread-safe FIFO queue for multiple processes | Many workers |
multiprocessing.Pipe |
Direct communication between two processes | 1-to-1 connection |
Manager / Value / Array |
Shared memory between processes | Small data, simple sync |
Socket / ZeroMQ |
Network communication between machines/processes | Large-scale distributed systems |
root_project/
│
├── workers/
│ └── client_worker.py # Single worker class running in separate process
│
├── utils/
│ └── global_worker.py # Main manager that creates, stores, and controls workers
│
├── controllers/
│ ├── GetMessageController.py # Example usage: fetch message
│ └── SendMessageController.py # Example usage: send message
│
├── main.py
├── README_ID.md
└── README.md
File: workers/client_worker.py
A subclass of multiprocessing.Process that runs as a separate process.
Its job is to read messages from queue_in, handle commands based on type, and return the result via queue_out.
from multiprocessing import Process, Queue
class ClientWorker(Process):
def __init__(self, queue_in, queue_out, worker_id):
super().__init__()
self.queue_in = queue_in
self.queue_out = queue_out
self.worker_id = worker_id
def run(self):
print(f"[Worker-{self.worker_id}] started ✅")
while True:
msg = self.queue_in.get()
if msg is None:
print(f"[Worker-{self.worker_id}] shutting down...")
break
msg_type = msg.get("type")
data = msg.get("data")
if msg_type == "get_message_byid":
print(f"[Worker-{self.worker_id}] Fetch message for chatId={data.get('chatId')}")
self.queue_out.put({
"from": self.worker_id,
"type": msg_type,
"result": "Fetched message"
})
elif msg_type == "send_message":
print(f"[Worker-{self.worker_id}] Sending message:", data)
self.queue_out.put({
"from": self.worker_id,
"type": msg_type,
"result": "Message sent"
})
else:
print(f"[Worker-{self.worker_id}] Unknown message type:", msg_type)File: utils/global_worker.py
Acts as a central manager that stores all worker instances (wa_1, wa_2, etc.) in a dictionary _instances.
Provides methods to:
- Create a worker
- Send message (
send) - Receive result (
receive) - Stop worker (
stop)
from multiprocessing import Queue
from workers.client_worker import ClientWorker
class ClientWorkerManager:
_instances = {}
@classmethod
def get(cls, worker_id):
return cls._instances.get(worker_id)
@classmethod
def create(cls, worker_id):
if worker_id in cls._instances:
print(f"{worker_id} already exist in instances")
return cls._instances[worker_id]
queue_in = Queue()
queue_out = Queue()
worker = ClientWorker(queue_in, queue_out, worker_id)
worker.start()
cls._instances[worker_id] = {
"queue_in": queue_in,
"queue_out": queue_out,
"process": worker
}
print(f"[Manager] Worker {worker_id} created.")
return cls._instances[worker_id]
@classmethod
def send(cls, worker_id, type_, data=None):
worker = cls.get(worker_id)
if not worker:
print(f"[Manager] ❌ Worker {worker_id} not found")
return None
msg = {"type": type_, "data": data or {}}
worker["queue_in"].put(msg)
print(f"[Manager] Sent to worker {worker_id}: {msg}")
return True
@classmethod
def receive(cls, worker_id):
worker = cls.get(worker_id)
if worker and not worker["queue_out"].empty():
return worker["queue_out"].get()
return None
@classmethod
def stop(cls, worker_id):
worker = cls.get(worker_id)
if worker:
worker["queue_in"].put(None)
worker["process"].join()
del cls._instances[worker_id]
print(f"[Manager] Worker {worker_id} stopped.")from utils.global_worker import ClientWorkerManager
import time
worker_id = "wa_1"
ClientWorkerManager.create(worker_id)
ClientWorkerManager.send(worker_id, "get_message_byid", {"chatId": "abc123"})
time.sleep(1)
resp = ClientWorkerManager.receive(worker_id)
print("Response A:", resp)from utils.global_worker import ClientWorkerManager
import time
worker_id = "wa_1"
ClientWorkerManager.create(worker_id)
ClientWorkerManager.send(worker_id, "send_message", {"text": "Hello from controller B!"})
time.sleep(1)
resp = ClientWorkerManager.receive(worker_id)
print("Response B:", resp)To make communication non-blocking (without time.sleep), use asyncio + thread executor:
import asyncio
async def send_and_wait(worker_id, type_, data):
ClientWorkerManager.send(worker_id, type_, data)
while True:
resp = ClientWorkerManager.receive(worker_id)
if resp:
return resp
await asyncio.sleep(0.05)✅ Process isolation — each WhatsApp host has its own workspace
✅ High scalability — dynamic worker creation
✅ Data safety between processes — no shared memory interference
✅ Fault tolerance — one worker crash doesn’t affect others
✅ Flexible structure — easy to extend new command types
| Aspect | Node.js (child_process.fork()) |
Python (multiprocessing) |
|---|---|---|
| Model | Single-threaded event-loop + child | True multi-process |
| Communication | .send() & .on("message") |
Queue or Pipe |
| Worker Scaling | Easy for multiple child processes | Requires queue management |
| Shared Memory | Not directly supported | Optional via Manager() |
This structure builds an IPC-based worker system that is:
- Safe and scalable
- Suitable for multi-instance systems
- Easy to integrate with controller or REST API layers
- Python 3.9+
- No external libraries required (
multiprocessingis built-in)
MIT License © 2025
Developed by Joel Binsar Jupiter
for experimental parallel processing and IPC in Python.