SMPD block getting stuck and not executing

Hi

I am currently trying to build a smpd block to allow for mulitple workers to run at the same time. One worker is collecting data from a sensor, one is acting as a buffer and the other is processing the data. Yesterday I thought I got it working and was working the way I wanted it too, but for some reason when I booted up matlab today and tried to run it again, it keeps getting stuck at the same place and not executing. I am not sure if it is something I did or if it is the software. Doesn't make sense that it is me as it was running fine yesterday and I have not changed anything, but I might have gotten lucky yesterday

This is where it gets stuck in the code:

This is the code:

%%create the pool
parpool(3);

%%Create the data queue 
Q = parallel.pool.DataQueue;
afterEach(Q,@update);

global sampleRate;
sampleRate = 51200;

%Time measurement
tic;

spmd
    switch spmdIndex
        
        %% Data Aquisition 
        case 1
            for i = 1:50
                %% Create DataAcquisition Object
                %Create a DataAcquisition object for the specified vendor.
                d = daq("ni");
               
                %% Set DataAcquisition Rate
                % Set scan rate.
                d.Rate = sampleRate;
                
                %% Add Channels
                %Add channels and set channel properties, if any.
                addinput(d,"cDAQ1Mod1","ai0","Voltage");
                
                addinput(d,"cDAQ1Mod1","ai1","Voltage");
                
                addinput(d,"cDAQ1Mod1","ai2","Voltage");
              
                %%% read data every 0.5secons
                n = ceil(d.Rate);
                data = read(d,n);
                
                stop(d) 
                
                %send data between workers
                sent = 1;

                if ~isempty(data)
                    spmdSend(sent,3,2) %check that the data has been read correctly
                    spmdSend(data,3,1)
                end
                
                %Reset the DAQ object 
                daqreset
            end

            %send the end message 
            runEnd = 2;
            spmdSend(runEnd, 3,2)
        
        %% Data processing 
        case 2
            
            while 1
                %send signal to say ready to process again

                rToP = 1;
                spmdSend(rToP,3,4)

                startP = spmdReceive(3,5);

                if startP == 1
                    
                    %send signal that it is not ready to receive any more
                    %data
                    rToP = 0;
                    spmdSend(rToP,3,4)
                    
                    %%Receive the data
                    data = spmdReceive(3,3);
                    
                    %% Perform FFT
                    cDAQ_ai0 = fft(data(:,1));
                    cDAQ_ai1 = fft(data(:,2));
                    cDAQ_ai2 = fft(data(:,3));
                    
                    
                    %%interperating the results 
                    cDAQ_ai0_p = real(cDAQ_ai0)';
                    cDAQ_ai1_p = real(cDAQ_ai1)';
                    cDAQ_ai2_p = real(cDAQ_ai2)';
            
                    send(Q, [cDAQ_ai0_p; cDAQ_ai1_p; cDAQ_ai2_p]);

                elseif startP == 2
                    break
                end
            end
        
        %% Buffer worker 
        case 3 
            
            %Create the buffer
            dataBuff = dsp.AsyncBuffer(200000); 
            dataRec = 0;
            workerRed = 0;
            
            while 1
                %check the state of the signal sent is
                startP = 0;
                sent = spmdReceive(1,2);
                dataRec = sent;
                
                %run if signal has been recieved from worker 1
                if dataRec == 1
                    %Write data from the buffer
                    data = spmdReceive(1,1);
                    dToB = [data.cDAQ1Mod1_ai0 data.cDAQ1Mod1_ai1 data.cDAQ1Mod1_ai2];
                    write(dataBuff, dToB); %write into buffer
                elseif dataRec == 2
                    break
                end
                %check if the end signal has been sent
                
                %Send data to worker 2 if the ready to process signal is
                %sent
                rToP = spmdReceive(2,4);
                if rToP == 1
                    %Sending data to process worker
                    dToP = read(dataBuff,sampleRate);
                    spmdSend(dToP,2,3)
                    
                    %Send message to start processing
                    startP = 1;
                    spmdSend(startP,2,5)

                end
                
                %hold the start of processing 
                holdP = 0;
                spmdSend(holdP,2,5)
            end
            
            %stop processing at the end of the loop 
            stopP = 2;
            spmdSend(stopP,2,5)
    end
end


%% Update the plots 

function update(d)
 
    time_p = linspace(1.9531e-05, 1, 51197);

    figure(1)
    plot(time_p,d(1,3:end-1),time_p,d(2,3:end-1),time_p,d(3,3:end-1))

    legend('X','Y','Z')

    fprintf('\nprocessing done\n')
    toc

end

Anyhelp anyone is able to give would be amazing. Thanks!

Answer 1

[Edric Ellis]

I suspect the problem here is that the way that you're using spmdSend and spmdReceive doesn't quite match how they work. When you call spmdReceive, it always receives the next message in turn, and blocks until the message is there. All messages that are sent will be received. Your code seems to me to be expecting spmdReceive to pick up only the most-recently-sent message. (Also, you're using the tag parameter, so the receiver will wait for the sender to send a message with that precise tag).

The factor that often catches people out is that spmdSend also can block - sometimes. Precisely when depends on the underlying MPI library used to implement spmdSend. Typically, for tiny messages, spmdSend completes immediately; somewhere around 100-200kB (I think), spmdSend starts to block until the matching spmdReceive has started running.

So, I think you probably need to re-work your messaging - especially between workers 2 and 3. I think what you're trying to achieve is to let worker 3 buffer data while worker 2 is processing, and then offload all the buffered data when worker 2 has finished the previous batch. This is not a completely natural fit for MPI-style programming, but I think it can be done by using spmdProbe.

You can use mpiSettings to turn on logging to the command-window, which might help you work out where things are getting stuck. The error stack you posted doesn't tell you much - that just tells you where the MATLAB client was when things got stuck, and that stack means simply "I am waiting for an spmd block to complete".

Here's a quick sketch of how I think I would go about this.

spmd
    if spmdIndex == 1
        for i = 1:50
            data = readSomeData();
            if ~isempty(data)
                spmdSend(data, 3, 1); % A data message
            end
        end
        spmdSend(true, 3, 2); % Tag 2 means "finished"
    elseif spmdIndex == 2
        while true
            spmdSend(true, 3) % "Ready to process"
            [data, ~, tag] = spmdReceive(3);
            if tag == 2
                % "Finish" tag
                break
            end
            doStuff(data);
        end
    elseif spmdIndex == 3
        keepReceiving = true;
        while keepReceiving
            [data, ~, tag] = spmdReceive(1);
            if tag == 2
                % "Finish" tag - nothing more to receive
                keepReceiving
            else
                % Put data in the buffer
            end
            % Check if worker 2 is ready
            worker2IsReady = spmdProbe(2);
            if worker2IsReady
                spmdReceive(2); % Drain the message
                spmdSend(data, 2);
            end
        end
        % We have finished. We need to wait for worker 2 to be ready again
        % and then we tell it to finish
        spmdReceive(2);
        spmdSend(true, 2, 2);
    end
end

[Michael Romeo]

Thankyou for that, that makes a lot of sense. I think that has cleared up the blockage. So how exactly does the receive, send and probe method work. If there is no recent message being sent, that hasn't already been read, will a worker stay on the line of code with the spmdReceive? And does probe simply return true if the code is on the receive line? Or is that true coming from the spmdSend just above?

[Edric Ellis]

Yes, you're about right there. spmdReceive always blocks until a message is ready. spmdProbe tells you if a call to spmdReceive will proceed "immediately" (i.e. there's data ready to receive, but there may still be some work to do to fully receive it). "Recent" doesn't really come into things - messages are always received in the order they were sent (providing they match your constraints in terms of tag and source).

For spmdSend and spmdReceive, you can think of each worker as having a dedicated "receive queue". When another worker sends a message, it ends up in the destination receive queue. When spmdSend has to block (because the message is large), then (effectively) the entry in the "receive queue" indicates that the receiver needs to pull the rest of the data across from the sender. When you call spmdReceive, the oldest matching message in the "receive queue" is returned - if nothing matches, then the call blocks. "Matching" is based on source + tag. spmdProbe effectively tells you whether a matching message is already present in the queue.

[Michael Romeo]

That makes sense thankyou. I am having another issue at the moment that doesn't make much sense. So the end statement in which the tag is changed from 1 to 2 and data is given the value of true to stop the while loops is for some reason not sending. So 'data' is changing to the logical true, but the tag is not changing, it is staying as 1, which is faulting out the code. Do you know why this might be happening?

[Edric Ellis]

Doh, that's because I got the syntax wrong! The 2nd output of spmdReceive is the source worker index, the 3rd output is the tag value. I'll update the code. Sorry about that. That's what comes of not testing stuff before posting it!

[Michael Romeo]

No that is absolutely no worries, you have helped me plenty