JVM Low-Level I/O - Part 4

IPC is how separate processes exchange data. On the JVM, we have access to surprisingly powerful IPC mechanisms — from memory-mapped files (the fastest) to Unix domain sockets (the most flexible). This article teaches you when and how to use each one.

---

1. What is IPC?

Inter-Process Communication (IPC) is any mechanism that allows separate processes (separate JVM instances, or Java ↔ native code) to exchange data.

Why Not Just Use Threads?

Great question! If you can use threads in a single JVM, do that — it's simpler. But IPC is unavoidable when:

Scenario	Why IPC?
Microservices	Each service is a separate process
Language boundary	Java talking to C++/Rust/Python
Process isolation	Security, fault isolation, different JVM versions
Crash resilience	One process crashing shouldn't kill others
Resource limits	Separate heap sizes, GC configurations

---

2. IPC Mechanisms Overview

Decision Matrix

---

3. Shared Memory via Memory-Mapped Files

This is the fastest IPC mechanism available on the JVM. Two (or more) processes map the same file into their virtual address spaces. Writes by one process are instantly visible to the other — no system calls, no copies, no kernel involvement for data transfer.

Architecture

Key insight: Both processes are reading from and writing to the same physical memory pages. The "file" is just an anchor point — for shared memory IPC, you can use a tmpfs/ramfs file so no disk is ever involved.

Producer Process

import java.nio.*;
import java.nio.channels.*;
import java.nio.file.*;

/**
 * Producer: Writes messages to shared memory.
 * 
 * Shared memory layout:
 * ┌───────────┬────────────┬─────────────────────┐
 * │ ready (4) │ length (4) │ payload (up to 1016) │
 * └───────────┴────────────┴─────────────────────┘
 * Total: 1024 bytes
 */
public class SharedMemProducer {
    
    private static final int SHM_SIZE = 1024;
    private static final int READY_OFFSET = 0;
    private static final int LENGTH_OFFSET = 4;
    private static final int PAYLOAD_OFFSET = 8;
    
    public static void main(String[] args) throws Exception {
        Path shmPath = Path.of(System.getProperty("java.io.tmpdir"), "ipc_shm.dat");
        
        try (FileChannel channel = FileChannel.open(shmPath,
                StandardOpenOption.READ,
                StandardOpenOption.WRITE,
                StandardOpenOption.CREATE)) {
            
            // Ensure file is the right size
            if (channel.size() < SHM_SIZE) {
                channel.write(ByteBuffer.allocate(SHM_SIZE), 0);
            }
            
            MappedByteBuffer shm = channel.map(
                FileChannel.MapMode.READ_WRITE, 0, SHM_SIZE);
            
            // Send 10 messages
            for (int i = 0; i < 10; i++) {
                // Wait until consumer marks buffer as not-ready
                while (shm.getInt(READY_OFFSET) == 1) {
                    Thread.onSpinWait(); // CPU-friendly spin
                }
                
                // Write message
                String message = "Message #" + i + " at " + System.nanoTime();
                byte[] payload = message.getBytes(java.nio.charset.StandardCharsets.UTF_8);
                
                shm.putInt(LENGTH_OFFSET, payload.length);
                shm.position(PAYLOAD_OFFSET);
                shm.put(payload);
                
                // Memory barrier: ensure payload is visible before ready flag
                // VarHandle or Unsafe would be more correct, but force() suffices 
                // for cross-process visibility
                shm.force();
                
                // Signal ready
                shm.putInt(READY_OFFSET, 1);
                shm.force();
                
                System.out.println("Sent: " + message);
                Thread.sleep(100);
            }
        }
    }
}

Consumer Process

public class SharedMemConsumer {
    
    private static final int SHM_SIZE = 1024;
    private static final int READY_OFFSET = 0;
    private static final int LENGTH_OFFSET = 4;
    private static final int PAYLOAD_OFFSET = 8;
    
    public static void main(String[] args) throws Exception {
        Path shmPath = Path.of(System.getProperty("java.io.tmpdir"), "ipc_shm.dat");
        
        try (FileChannel channel = FileChannel.open(shmPath,
                StandardOpenOption.READ,
                StandardOpenOption.WRITE)) {
            
            MappedByteBuffer shm = channel.map(
                FileChannel.MapMode.READ_WRITE, 0, SHM_SIZE);
            
            int messagesReceived = 0;
            while (messagesReceived < 10) {
                // Spin until producer signals ready
                while (shm.getInt(READY_OFFSET) != 1) {
                    Thread.onSpinWait();
                }
                
                // Read message
                int length = shm.getInt(LENGTH_OFFSET);
                byte[] payload = new byte[length];
                shm.position(PAYLOAD_OFFSET);
                shm.get(payload);
                
                String message = new String(payload, 
                    java.nio.charset.StandardCharsets.UTF_8);
                System.out.println("Received: " + message);
                
                // Mark as consumed
                shm.putInt(READY_OFFSET, 0);
                shm.force();
                
                messagesReceived++;
            }
        }
    }
}

Running the Example

# Terminal 1 — Start consumer first
java SharedMemConsumer

# Terminal 2 — Start producer
java SharedMemProducer

---

4. Pipes (Stdin/Stdout IPC)

The simplest form of IPC. A parent process launches a child and communicates through stdin/stdout streams. Limited to parent-child relationships.

Parent Process

public class PipeParent {
    public static void main(String[] args) throws Exception {
        ProcessBuilder pb = new ProcessBuilder("java", "PipeChild");
        pb.redirectErrorStream(true); // merge stderr into stdout
        
        Process child = pb.start();
        
        // Write to child's stdin
        OutputStream childStdin = child.getOutputStream();
        BufferedWriter writer = new BufferedWriter(
            new OutputStreamWriter(childStdin, StandardCharsets.UTF_8));
        
        // Read from child's stdout
        InputStream childStdout = child.getInputStream();
        BufferedReader reader = new BufferedReader(
            new InputStreamReader(childStdout, StandardCharsets.UTF_8));
        
        // Send command
        writer.write("PROCESS 42");
        writer.newLine();
        writer.flush();
        
        // Read response
        String response = reader.readLine();
        System.out.println("Child responded: " + response);
        
        // Cleanup
        writer.write("EXIT");
        writer.newLine();
        writer.flush();
        child.waitFor();
    }
}

Child Process

public class PipeChild {
    public static void main(String[] args) throws Exception {
        BufferedReader stdin = new BufferedReader(
            new InputStreamReader(System.in, StandardCharsets.UTF_8));
        PrintWriter stdout = new PrintWriter(
            new OutputStreamWriter(System.out, StandardCharsets.UTF_8), true);
        
        String line;
        while ((line = stdin.readLine()) != null) {
            if (line.equals("EXIT")) break;
            
            if (line.startsWith("PROCESS")) {
                int value = Integer.parseInt(line.split(" ")[1]);
                stdout.println("RESULT " + (value * 2));
            }
        }
    }
}

Using Pipe with ByteBuffer (NIO Channels)

// Convert stream to channel for ByteBuffer usage
Process child = pb.start();
WritableByteChannel toChild = Channels.newChannel(child.getOutputStream());
ReadableByteChannel fromChild = Channels.newChannel(child.getInputStream());

ByteBuffer buffer = ByteBuffer.allocate(1024);

// Write structured data
buffer.putInt(42);
buffer.putDouble(3.14);
buffer.flip();
toChild.write(buffer);

// Read response
buffer.clear();
fromChild.read(buffer);
buffer.flip();
int result = buffer.getInt();

---

5. Unix Domain Sockets (Java 16+)

Unix Domain Sockets are like TCP sockets but for same-machine communication. They're faster than TCP because they skip the network stack (no checksums, no routing, no packet fragmentation).

Server

import java.net.UnixDomainSocketAddress;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;

public class UDSServer {
    public static void main(String[] args) throws Exception {
        Path socketPath = Path.of(System.getProperty("java.io.tmpdir"), "ipc.sock");
        Files.deleteIfExists(socketPath);
        
        UnixDomainSocketAddress address = UnixDomainSocketAddress.of(socketPath);
        
        try (ServerSocketChannel server = ServerSocketChannel.open(
                java.net.StandardProtocolFamily.UNIX)) {
            
            server.bind(address);
            System.out.println("Server listening on: " + socketPath);
            
            try (SocketChannel client = server.accept()) {
                System.out.println("Client connected!");
                
                ByteBuffer buffer = ByteBuffer.allocate(1024);
                
                // Read request
                client.read(buffer);
                buffer.flip();
                
                int requestType = buffer.getInt();
                int value = buffer.getInt();
                System.out.printf("Request: type=%d, value=%d%n", requestType, value);
                
                // Send response
                buffer.clear();
                buffer.putInt(0); // status: OK
                buffer.putInt(value * 2);  // result
                buffer.flip();
                client.write(buffer);
            }
        } finally {
            Files.deleteIfExists(socketPath);
        }
    }
}

Client

public class UDSClient {
    public static void main(String[] args) throws Exception {
        Path socketPath = Path.of(System.getProperty("java.io.tmpdir"), "ipc.sock");
        UnixDomainSocketAddress address = UnixDomainSocketAddress.of(socketPath);
        
        try (SocketChannel channel = SocketChannel.open(
                java.net.StandardProtocolFamily.UNIX)) {
            
            channel.connect(address);
            System.out.println("Connected to server");
            
            ByteBuffer buffer = ByteBuffer.allocate(1024);
            
            // Send request
            buffer.putInt(1);   // request type
            buffer.putInt(21);  // value
            buffer.flip();
            channel.write(buffer);
            
            // Read response
            buffer.clear();
            channel.read(buffer);
            buffer.flip();
            
            int status = buffer.getInt();
            int result = buffer.getInt();
            System.out.printf("Response: status=%d, result=%d%n", status, result);
            // Output: Response: status=0, result=42
        }
    }
}

---

6. TCP Socket IPC

When you need IPC over a network, or maximum portability, use TCP sockets with NIO:

import java.nio.channels.*;
import java.net.*;

public class NIOTCPServer {
    public static void main(String[] args) throws Exception {
        Selector selector = Selector.open();
        
        ServerSocketChannel server = ServerSocketChannel.open();
        server.bind(new InetSocketAddress("localhost", 9876));
        server.configureBlocking(false);
        server.register(selector, SelectionKey.OP_ACCEPT);
        
        ByteBuffer buffer = ByteBuffer.allocate(1024);
        
        System.out.println("Server listening on port 9876");
        
        while (true) {
            selector.select(); // blocks until events
            
            var keys = selector.selectedKeys().iterator();
            while (keys.hasNext()) {
                SelectionKey key = keys.next();
                keys.remove();
                
                if (key.isAcceptable()) {
                    // New connection
                    SocketChannel client = server.accept();
                    client.configureBlocking(false);
                    client.register(selector, SelectionKey.OP_READ);
                    System.out.println("Client connected: " + 
                        client.getRemoteAddress());
                    
                } else if (key.isReadable()) {
                    // Data available
                    SocketChannel client = (SocketChannel) key.channel();
                    buffer.clear();
                    int bytesRead = client.read(buffer);
                    
                    if (bytesRead == -1) {
                        client.close();
                        continue;
                    }
                    
                    buffer.flip();
                    // Echo back
                    client.write(buffer);
                }
            }
        }
    }
}

---

7. Comparing IPC Methods

Method	Latency	Throughput	Complexity	Cross-machine	JVM Version
Shared Memory	~50-100ns	Excellent	High	❌	Any
Unix Domain Socket	~1-3μs	Very Good	Medium	❌	16+
Pipe (stdin/stdout)	~1-5μs	Good	Low	❌	Any
TCP Socket	~5-50μs	Good	Medium	✅	Any
File I/O	~100μs+	Poor	Low	❌*	Any

*Can work across NFS, but very slow

---

8. Building a Shared Memory IPC System

Let's build a more robust shared memory IPC system with a proper header, sequence numbers, and a message protocol.

Memory Layout

Complete Implementation

import java.nio.*;
import java.nio.channels.*;
import java.nio.charset.StandardCharsets;
import java.nio.file.*;

/**
 * Robust shared memory IPC with header, sequence numbers, and CRC.
 */
public class SharedMemoryIPC {
    
    // Header layout
    private static final int MAGIC = 0xDEADBEEF;
    private static final int VERSION = 1;
    
    private static final int OFFSET_MAGIC = 0;
    private static final int OFFSET_VERSION = 4;
    private static final int OFFSET_PRODUCER_SEQ = 8;
    private static final int OFFSET_CONSUMER_SEQ = 16;
    private static final int OFFSET_MSG_SIZE = 24;
    private static final int OFFSET_RESERVED = 28;
    private static final int HEADER_SIZE = 32;
    
    private static final int SHM_SIZE = 4096;
    private static final int MAX_MSG_SIZE = SHM_SIZE - HEADER_SIZE;
    
    private final MappedByteBuffer shm;
    private final FileChannel channel;
    
    public SharedMemoryIPC(Path path, boolean create) throws Exception {
        if (create) {
            this.channel = FileChannel.open(path,
                StandardOpenOption.READ,
                StandardOpenOption.WRITE,
                StandardOpenOption.CREATE);
            
            // Initialize the file
            if (channel.size() < SHM_SIZE) {
                ByteBuffer zeros = ByteBuffer.allocate(SHM_SIZE);
                channel.write(zeros, 0);
            }
            
            this.shm = channel.map(FileChannel.MapMode.READ_WRITE, 0, SHM_SIZE);
            
            // Write header
            shm.putInt(OFFSET_MAGIC, MAGIC);
            shm.putInt(OFFSET_VERSION, VERSION);
            shm.putLong(OFFSET_PRODUCER_SEQ, 0);
            shm.putLong(OFFSET_CONSUMER_SEQ, 0);
            shm.force();
            
        } else {
            this.channel = FileChannel.open(path,
                StandardOpenOption.READ,
                StandardOpenOption.WRITE);
            this.shm = channel.map(FileChannel.MapMode.READ_WRITE, 0, SHM_SIZE);
            
            // Validate header
            int magic = shm.getInt(OFFSET_MAGIC);
            if (magic != MAGIC) {
                throw new IllegalStateException(
                    "Invalid shared memory: bad magic 0x" + Integer.toHexString(magic));
            }
        }
    }
    
    /**
     * Send a message (producer side).
     * Blocks until consumer has consumed the previous message.
     */
    public void send(byte[] data) {
        if (data.length > MAX_MSG_SIZE) {
            throw new IllegalArgumentException(
                "Message too large: " + data.length + " > " + MAX_MSG_SIZE);
        }
        
        long producerSeq = shm.getLong(OFFSET_PRODUCER_SEQ);
        long consumerSeq;
        
        // Wait until consumer catches up (at most 1 message ahead)
        do {
            consumerSeq = shm.getLong(OFFSET_CONSUMER_SEQ);
            if (producerSeq > consumerSeq) {
                Thread.onSpinWait();
            }
        } while (producerSeq > consumerSeq);
        
        // Write message
        shm.putInt(OFFSET_MSG_SIZE, data.length);
        for (int i = 0; i < data.length; i++) {
            shm.put(HEADER_SIZE + i, data[i]);
        }
        
        // Publish: increment sequence number
        shm.putLong(OFFSET_PRODUCER_SEQ, producerSeq + 1);
        shm.force();
    }
    
    /**
     * Receive a message (consumer side).
     * Blocks until a new message is available.
     */
    public byte[] receive() {
        long consumerSeq = shm.getLong(OFFSET_CONSUMER_SEQ);
        long producerSeq;
        
        // Wait for new message
        do {
            producerSeq = shm.getLong(OFFSET_PRODUCER_SEQ);
            if (producerSeq <= consumerSeq) {
                Thread.onSpinWait();
            }
        } while (producerSeq <= consumerSeq);
        
        // Read message
        int size = shm.getInt(OFFSET_MSG_SIZE);
        byte[] data = new byte[size];
        for (int i = 0; i < size; i++) {
            data[i] = shm.get(HEADER_SIZE + i);
        }
        
        // Acknowledge: increment consumer sequence
        shm.putLong(OFFSET_CONSUMER_SEQ, consumerSeq + 1);
        shm.force();
        
        return data;
    }
    
    public void close() throws Exception {
        channel.close();
    }
    
    // --- Helper methods for sending/receiving strings ---
    
    public void sendString(String msg) {
        send(msg.getBytes(StandardCharsets.UTF_8));
    }
    
    public String receiveString() {
        return new String(receive(), StandardCharsets.UTF_8);
    }
}

Usage — Producer

public class IPCProducerApp {
    public static void main(String[] args) throws Exception {
        Path shmPath = Path.of(System.getProperty("java.io.tmpdir"), "robust_ipc.shm");
        SharedMemoryIPC ipc = new SharedMemoryIPC(shmPath, true); // create
        
        for (int i = 0; i < 1_000_000; i++) {
            ipc.sendString("Message #" + i);
            if (i % 100_000 == 0) {
                System.out.println("Sent: " + i);
            }
        }
        
        ipc.sendString("__EXIT__");
        ipc.close();
    }
}

Usage — Consumer

public class IPCConsumerApp {
    public static void main(String[] args) throws Exception {
        Path shmPath = Path.of(System.getProperty("java.io.tmpdir"), "robust_ipc.shm");
        
        // Wait for producer to create the file
        while (!Files.exists(shmPath)) {
            Thread.sleep(100);
        }
        
        SharedMemoryIPC ipc = new SharedMemoryIPC(shmPath, false); // open existing
        
        long start = System.nanoTime();
        int count = 0;
        
        while (true) {
            String msg = ipc.receiveString();
            if ("__EXIT__".equals(msg)) break;
            count++;
        }
        
        long elapsed = System.nanoTime() - start;
        System.out.printf("Received %,d messages in %,d ms%n", 
            count, elapsed / 1_000_000);
        System.out.printf("Throughput: %,.0f messages/sec%n", 
            count / (elapsed / 1_000_000_000.0));
        
        ipc.close();
    }
}

---

9. Synchronization Challenges

Shared memory IPC is fast but comes with synchronization challenges:

The Visibility Problem

Problem: CPU caches may hold stale data. The consumer might see the ready flag change before the payload data is visible!

Solution: Memory Barriers

import java.lang.invoke.VarHandle;

// For intra-JVM shared memory (same JVM, different threads):
// Use VarHandle for proper memory ordering

// For inter-JVM shared memory (different processes):
// Use MappedByteBuffer.force() or volatile semantics

// The sequence number pattern we used above is a simple form of this:
// 1. Producer writes data
// 2. Producer calls force() (memory barrier)
// 3. Producer increments sequence number
// 4. Consumer reads sequence number
// 5. Consumer reads data (guaranteed to see producer's writes)

Ordering Solutions Comparison

---

10. Real-World IPC Architecture

Here's an architecture for a high-performance system using multiple IPC channels:

Architecture decisions:

• Shared memory for the hottest path (market data) → lowest latency

• Unix domain sockets for command/control → bidirectional, reliable

• TCP sockets for external exchange connectivity → required by protocol

---

11. Summary

Key takeaways:

1. Shared memory (via MappedByteBuffer) is the fastest IPC — sub-microsecond

2. Unix Domain Sockets balance speed and convenience for same-machine IPC

3. Pipes are simple but limited to parent-child process relationships

4. TCP sockets are required for cross-machine communication

5. Synchronization is the hardest part of shared memory IPC

6. Choose IPC method based on latency, throughput, and simplicity needs