Memory API

Memory regions provide high-throughput shared memory storage optimized for large data structures that need to be accessed and modified by multiple clients across the cluster. Unlike key/value operations, memory regions support direct byte-level access at specific offsets for maximum performance.

Features

• High Throughput: Optimized for large data transfers with gRPC streaming
• Direct Access: Read and write at specific byte offsets without full data transfer
• Atomic Operations: Individual write operations are atomic within their byte range
• NVMe-oF Backend: Optional dataplane service for sub-millisecond access latency
• Dynamic Sizing: Resize regions as data requirements change
• Namespace Isolation: Complete tenant separation with per-namespace access control
• Concurrent Access: Multiple clients can access different ranges simultaneously

allocate_memory

Reserves a named region with a given size in bytes. Returns ALREADY_EXISTS if the region is present.

message AllocateMemoryRequest {
  string key = 1;
  string namespace = 2;
  uint64 size_bytes = 3;
}

message AllocateMemoryResponse {
  string key = 1;
  string namespace = 2;
  uint64 size_bytes = 3;
  Status status = 4;
}

write_memory

Writes bytes at an offset within the region. The call is atomic for the provided byte range. A timeout_ms can be supplied for large transfers to avoid indefinite blocking.

message WriteMemoryRequest {
  string key = 1;
  string namespace = 2;
  uint64 offset = 3;
  bytes data = 4;
  optional uint64 timeout_ms = 5;  // Time to wait for the write in milliseconds
}

message WriteMemoryResponse {
  string key = 1;
  string namespace = 2;
  uint64 written_bytes = 3;
  Status status = 4;
}

Automatic Region Growth

If a write extends beyond the current capacity, the region automatically resizes to offset + len(data). data_length is updated accordingly.

read_memory

Reads bytes at an offset within the region. An OUT_OF_RANGE error is returned if the requested range exceeds the region size.

ReadMemoryRequest {
  namespace: string
  name: string
  offset: uint64
  length: uint64
  timeout_ms: uint64
}

resize_memory

Changes the total size of the region. Existing data is preserved when growing. Shrinking truncates bytes beyond the new size.

ResizeMemoryRequest {
  namespace: string
  name: string
  new_size: uint64
  timeout_ms: uint64
}

free_memory

Deletes the region and releases its resources. Returns NOT_FOUND if the region does not exist.

FreeMemoryRequest {
  namespace: string
  name: string
  timeout_ms: uint64
}

describe_memory

Returns metadata about a region including its allocated capacity and the length of the data written so far. The call accepts an optional timeout.

DescribeMemoryRequest {
  namespace: string
  name: string
  timeout_ms: uint64
}

list_memory_regions

Lists the keys of memory regions in the namespace. A prefix can be supplied to filter the results.

ListMemoryRegionsRequest {
  namespace: string
  prefix: string
}

CLI usage:

primatomic-cli memory list <namespace> [prefix]

SDK Examples

Python SDK - Basic Operations

import primatomic
import struct

client = primatomic.Client("localhost:50051", credential="your-jwt-token")

# Allocate a 1MB memory region
region_id = await client.allocate_memory("myapp", "shared_buffer", size=1024*1024)

# Write structured data at specific offsets
header_data = struct.pack('!I', 42)  # 4-byte integer
await client.write_memory("myapp", "shared_buffer", offset=0, data=header_data)

payload = b"Hello, distributed world!"
await client.write_memory("myapp", "shared_buffer", offset=4, data=payload)

# Read back the data
header_bytes = await client.read_memory("myapp", "shared_buffer", offset=0, length=4)
header_value = struct.unpack('!I', header_bytes)[0]

payload_bytes = await client.read_memory("myapp", "shared_buffer", offset=4, length=25)
print(f"Header: {header_value}, Payload: {payload_bytes.decode()}")

# Get region metadata
info = await client.describe_memory("myapp", "shared_buffer")
print(f"Region size: {info.size} bytes, used: {info.data_length} bytes")

Large Data Transfer

# Efficient handling of large datasets
async def upload_large_file(file_path, region_name):
    file_size = os.path.getsize(file_path)

    # Allocate region to fit the file
    await client.allocate_memory("myapp", region_name, size=file_size)

    # Stream file in chunks for memory efficiency
    chunk_size = 64 * 1024  # 64KB chunks
    offset = 0

    with open(file_path, 'rb') as f:
        while True:
            chunk = f.read(chunk_size)
            if not chunk:
                break

            await client.write_memory("myapp", region_name, offset, chunk, timeout_ms=30000)
            offset += len(chunk)
            print(f"Uploaded {offset}/{file_size} bytes")

# Download large data efficiently
async def download_region(region_name, output_path):
    info = await client.describe_memory("myapp", region_name)
    chunk_size = 64 * 1024

    with open(output_path, 'wb') as f:
        offset = 0
        while offset < info.data_length:
            chunk_len = min(chunk_size, info.data_length - offset)
            chunk = await client.read_memory("myapp", region_name, offset, chunk_len)
            f.write(chunk)
            offset += len(chunk)

Dynamic Data Structures

# Implement a shared circular buffer
class SharedCircularBuffer:
    def __init__(self, client, namespace, name, size):
        self.client = client
        self.namespace = namespace
        self.name = name
        self.size = size
        self.header_size = 16  # 4 bytes each: write_pos, read_pos, count, capacity

    async def initialize(self):
        total_size = self.header_size + self.size
        await self.client.allocate_memory(self.namespace, self.name, total_size)

        # Initialize header: write_pos=0, read_pos=0, count=0, capacity=size
        header = struct.pack('!IIII', 0, 0, 0, self.size)
        await self.client.write_memory(self.namespace, self.name, 0, header)

    async def push(self, data):
        # Read current header
        header_bytes = await self.client.read_memory(self.namespace, self.name, 0, 16)
        write_pos, read_pos, count, capacity = struct.unpack('!IIII', header_bytes)

        if count >= capacity:
            raise BufferFull("Circular buffer is full")

        # Write data at current write position
        data_offset = self.header_size + write_pos
        await self.client.write_memory(self.namespace, self.name, data_offset, data)

        # Update header with new write position and count
        new_write_pos = (write_pos + len(data)) % capacity
        new_header = struct.pack('!IIII', new_write_pos, read_pos, count + 1, capacity)
        await self.client.write_memory(self.namespace, self.name, 0, new_header)

Region Management

# List all memory regions with filtering
regions = await client.list_memory_regions("myapp", prefix="cache:")
for region in regions:
    info = await client.describe_memory("myapp", region)
    print(f"Region {region}: {info.size} bytes allocated, {info.data_length} bytes used")

# Resize region dynamically
current_info = await client.describe_memory("myapp", "growing_dataset")
if current_info.data_length > current_info.size * 0.8:  # 80% full
    new_size = current_info.size * 2  # Double the size
    await client.resize_memory("myapp", "growing_dataset", new_size)
    print(f"Resized region to {new_size} bytes")

# Cleanup unused regions
await client.free_memory("myapp", "temporary_buffer")

Performance Optimization

Chunked Operations

# Optimal chunk sizes for different use cases
SMALL_CHUNK = 4 * 1024      # 4KB - good for frequent small updates
MEDIUM_CHUNK = 64 * 1024    # 64KB - balanced for most operations
LARGE_CHUNK = 1024 * 1024   # 1MB - maximum throughput for bulk transfers

async def optimized_write(region_name, data, chunk_size=MEDIUM_CHUNK):
    total_size = len(data)
    tasks = []

    for offset in range(0, total_size, chunk_size):
        chunk = data[offset:offset + chunk_size]
        task = client.write_memory("myapp", region_name, offset, chunk)
        tasks.append(task)

    # Write chunks concurrently for maximum throughput
    await asyncio.gather(*tasks)

Error Handling

Common memory API error conditions:

• ALREADY_EXISTS: Region allocation failed because name is already used
• NOT_FOUND: Operation on non-existent region
• OUT_OF_RANGE: Read/write offset exceeds region boundaries
• DEADLINE_EXCEEDED: Operation timeout on large transfers
• RESOURCE_EXHAUSTED: Insufficient space for allocation or resize
• PERMISSION_DENIED: Invalid credential ID or insufficient namespace access

Best Practices

Design Guidelines

• Region Sizing: Allocate regions with growth headroom to minimize resizing
• Chunked Access: Use appropriate chunk sizes for your access patterns
• Offset Management: Track data layout carefully to avoid overlapping writes
• Timeout Configuration: Set generous timeouts for large data operations

Performance Optimization

• Concurrent Operations: Use parallel reads/writes for non-overlapping ranges
• NVMe-oF Deployment: Consider dataplane service for ultra-low latency requirements
• Region Pooling: Reuse regions instead of frequent allocation/deallocation
• Compression: Compress data before storage for regions with repetitive content

Operational Considerations

• Monitoring: Track region utilization and access patterns
• Cleanup Strategy: Implement region lifecycle management
• Backup Planning: Include memory regions in disaster recovery procedures