Passing output from afterEach to variable

Hello,

I`m trying to use parfeval that is running in the background to get images from multiple cameras and process them.

I tried using parallel.pool.DataQueue and parallel.pool.PollableDataQueue to get data from parfeval, but data being sent is too big, slowing down FPS and quickly filling up available ram.

My questions are:

1. Is there a way to fetch data on request instead of creating queue (I need only newest data - not sending data from worker after each iteration to queue)? The data can be inside worker untill it is needed.
2. Can queue be limited to size 1?
3. How to use afterEach to save data to variable? I can only find examples with disp and replacing data on figures, not saving to workspace.
4. Alternativly, is there a way to use timers in different threads instead of parfeval?

Thank you for your time,

Matt

[Walter Roberson]

3.

afterEach(queue,@foo)

function foo(data)
  assignin('base', 'data', data)
end

[Matt Tejer]

Thank you @Walter Roberson so much for your help. Do you perhaps know the answer to the last question about timers in different threads?

[Walter Roberson]

I think it would depend on which kind of camera it was, and which interface you are using. usb webcam support? ip camera support? Image Acquisition Toolbox support? GigE camera?

[Matt Tejer]

Intel RealSense d435i using USB interface with their wrapper for Matlab.

[Walter Roberson]

Are you using usb webcam support package or Image Acquisition toolbox?

[Matt Tejer]

As far as I am aware, none. https://github.com/IntelRealSense/librealsense/tree/master/wrappers/matlab

Example code

function depth_example()
% Make Pipeline object to manage streaming
pipe = realsense.pipeline();
% Make Colorizer object to prettify depth output
colorizer = realsense.colorizer();

% Start streaming on an arbitrary camera with default settings
profile = pipe.start();

% Get streaming device's name
dev = profile.get_device();
name = dev.get_info(realsense.camera_info.name);

% Get frames. We discard the first couple to allow
% the camera time to settle
for i = 1:5
    fs = pipe.wait_for_frames();
end

% Stop streaming
pipe.stop();

% Select depth frame
depth = fs.get_depth_frame();
% Colorize depth frame
color = colorizer.colorize(depth);

% Get actual data and convert into a format imshow can use
% (Color data arrives as [R, G, B, R, G, B, ...] vector)
data = color.get_data();
img = permute(reshape(data',[3,color.get_width(),color.get_height()]),[3 2 1]);

% Display image
imshow(img);
title(sprintf("Colorized depth frame from %s", name));
end

[Walter Roberson]

It is not clear to me that it would be beneficial to use that interface inside timer callbacks.

[Matt Tejer]

Timer gets the frames, processes them, sets timer.UserData that is available outside timer, waits for next frames. To me it seems better than parfeval where sending data from workers is slow. But creating timers in a "normal" way slows other timers down, being on the same thread. So my question is about creating parallel timers.

[Walter Roberson]

The documentation for timer has no section for extended capabilities, so there is no support for timers in thread pools.

I am not sure at the moment if there is support for timers in process pools. If there is, you would still have the same problem about returning the data to another process.

[Walter Roberson]

There is a faint chance that calling the external routines is compatible with thread pools, but I think it is unlikely.

Answer 1

[Walter Roberson]

This method is fairly clumsy, but in theory it would work.

Use a parpool size of 1.

Use parfeval() to create a parallel.pool.PollableDataQueue object and return it to the caller.

Create a parallel.pool.DataQueue object in the workspace.

parfeval() passing in the DataQueue object

afterEach() on the DataQueue

Now, when the client is ready to accept data, send() a signal (any value) from the client to the PollableDataQueue

Meanwhile, the worker is looping around fetching frames, and checking the QueueLength property of the PollableDataQueue. When the queue becomes non-empty, poll() the PollableDataQueue to "eat" the signal; then have the worker write the most recent frame to the DataQueue object. If the queue is empty, just loop back to accepting the next frame.

The afterEach function on the DataQueue receives the frame and stores it (and does whatever with it.) Oh, and the afterEach probably does the next send() to the PollableDataQueue to signal that the client is ready for more data.

Answer 2

[Thomas Falch]

Starting in R2025a you can use the new "any-destination" parallel.pool.PollableDataQueue to more easily send data from the client to workers (or from workers to workers). It is documented here: PollableDataQueue - Send and poll data between client and workers - MATLAB

You can either use it as a replacement for the queue created on the worker the second step in Walter's answer. An alternative solution could (particularily if you want to use more workers) could look like this:

% Runs on worker, creates data on client's request and sends it back
function createDataOnDemand(toClientQueue, toWorkerQueue)
while true
    % Wait for next request from client, or exit if client closes queue
    [dataParams, didReceive] = toWorkerQueue.poll(Inf);
    if ~didReceive
        break
    end
    % Generate next piece of data, and send back to client
    data = createData(dataParams);
    toClientQueue.send(data);
end
end

pool = parpool()

toClientQueue = parallel.pool.PollableDataQueue(Destination="any");
toWorkerQueue = parallel.pool.PollableDataQueue(Destination="any");

f = parfeval(@createDataOnDemand, toClientQueue, toWorkerQueue)

% Drive work from the client, send requests for data, then receive and
% process it
for i = 1:n
    dataParams = getDataParams(n);
    toWorkerQueue.send(dataParams);
    data = toClientQueue.poll(Inf);
    processData(data)
end

toWorkerQueue.close();