example_numpy.py

#!/usr/bin/env python3
"""Simple example with numpy arrays."""

import ctypes
import multiprocessing

import numpy
import numpy.matlib
import ringbuffer


def writer(ring):
    for i in range(10000):
        m = numpy.matlib.randn(25, 100)
        x = numpy.ctypeslib.as_ctypes(m)
        try:
            ring.try_write(x)
        except ringbuffer.WaitingForReaderError:
            print('Reader is too slow, dropping %r' % x)
            continue

        if i and i % 100 == 0:
            print('Wrote %d so far' % i)

    ring.writer_done()
    print('Writer is done')


def reader(ring, pointer):
    while True:
        try:
            data = ring.blocking_read(pointer)
        except ringbuffer.WriterFinishedError:
            break

        x = numpy.array(data[0])
        x.shape = (25, 100)
        x[1, 1] = 1.1  # Verify it's mutable

    print('Reader %r is done' % pointer)


def main():
    t = (ctypes.c_double * 100) * 25
    ring = ringbuffer.RingBuffer(c_type=t, slot_count=100)
    ring.new_writer()

    processes = [
        multiprocessing.Process(target=writer, args=(ring, )),
    ]
    for i in range(10):
        processes.append(
            multiprocessing.Process(
                target=reader, args=(ring, ring.new_reader())))

    for p in processes:
        p.start()

    for p in processes:
        print('join')
        p.join(timeout=50)
        assert not p.is_alive()
        assert p.exitcode == 0


if __name__ == '__main__':
    main()