Thread Subject: Parfor slices shared array. spmd copies. How to make a SHARED array?

Hello,

I need to access a shared matrix/array/vector/variable from multiple workers/labs and have the following problems:

x=zeros(10,1)
parfor i=1:length(x)
   if x(arbitraryIndex) == whatever
        x(i) = somecode....
    end
end

results in SLICING the 'x' vector and I cannot access all elements.

In:

spmd
    myIndexes = indexes(labindex,:);
    for i=myIndexes(1):myIndexes(2)
        % do whatever you desire with the array 'x'
    end
end

I get COPIES of x for every worker. In the final, the array 'x' is not the original matrix but an array of cells of size (num-cpu-cores (=workers)) which differs in the fields which were modified within body of a particular worker.

That means I cannot access the changes concurrently made by other workers. I need a REALLY SHARED array to be accessed at the same time from multiple workers (having in mind collisions may occur here, but that can be included into the plan and could be avoided by workers writing only into a separated segments of the array, however reading from the entire space).

Is there a way?

Thanks a lot for any tips!

Ibisek

I'm not familiar with the parallel toolbox, but I might suggest to take a look at this tool on FEX that might be of your interests.

http://www.mathworks.com/matlabcentral/fileexchange/24576

Bruno

"Bruno Luong" <b.luong@fogale.findmycountry> wrote in message <haf7db$avj$1@fred.mathworks.com>...
> I'm not familiar with the parallel toolbox, but I might suggest to take a look at this tool on FEX that might be of your interests.
>
> http://www.mathworks.com/matlabcentral/fileexchange/24576
>
> Bruno
Wow, very nice idea. When I experimented with the parallel toolbox a while ago, I hit this problem:

A = HugeStructureWithReadOnlyData;
parfor i=1:n
  x(i) = oper(A, i);
end

It resulted copying A to all workers (despite that it is read-only), leading to a serious memory problem.

What were your problems with structures?

Sebastiaan

"Sebastiaan " <s.breedveld@erasmusmc.REMOVE.BOO.BOO.nl> wrote in message <hafag6$okj$1@fred.mathworks.com>...
> "Bruno Luong" <b.luong@fogale.findmycountry> wrote in message <haf7db$avj$1@fred.mathworks.com>...
>
> What were your problems with structures?
>

Making similar function for structures and cells demand a serious investigation and reversed engineering on Matlab internal storing. I have spend a small amount of time, but did not get into the bottom.

Peter Boettcher has wrote a long post ago about such internal storing, unfortunately what he wrote is no longer valid for recent Matlab.

Bruno

"ibisek " <cervenka@fai.utb.cz> writes:

> I need to access a shared matrix/array/vector/variable from multiple
> workers/labs and have the following problems:
>
> [ ... parfor ... ]
>
> results in SLICING the 'x' vector and I cannot access all elements.
>
> In:
>
> spmd
> myIndexes = indexes(labindex,:);
> for i=myIndexes(1):myIndexes(2)
> % do whatever you desire with the array 'x'
> end
> end
>
> I get COPIES of x for every worker. In the final, the array 'x' is not the
>original matrix but an array of cells of size (num-cpu-cores (=workers)) which
>differs in the fields which were modified within body of a particular worker.

You need a (co)distributed array, where each worker stores just a portion of the
array. You can then work with the local parts on each worker. Here's an example
using R2009b syntax (note that there are some differences with

spmd
  x = codistributed.zeros( 100 ); % 100x100 array spread across the workers
  lp = getLocalPart( x ); % The underlying "local part"
  rr = globalIndices( x, 1 ); % Which rows in the global array do we have?
  cc = globalIndices( x, 2 ); % Which columns?
  for ii = 1:length( rr )
    for jj = 1:length( cc )
      % perform some calculation on the local part
      lp(ii, jj) = labindex * (rr(ii) + cc(jj));
    end
  end
  % We've modified the local part of "x", we need to put it back together
  x = codistributed.build( lp, getCodistributor( x ) );
end

% Back at the client, x is now a "distributed" array, with the data still on the
% workers. Use "gather" to bring it back to the client
xd = gather( x );

I must say that working directly with the local parts of codistributed arrays in
this way requires some care. In particular, you need to take care to call
"globalIndices" to find out which indices into the global array you have on each
worker, and when you've finished modifying the local part, you must use
"codistributed.build" to put things back together again. If you stick to the
methods defined for (co)distributed arrays - such as mtimes and so on - then the
communication is done for you "under the hood". Otherwise, there are labSend and
friends.

If you're not using R2009b, there are ways of doing all this stuff, but the
syntax is slightly different.

Cheers,

Edric.

Edric,

in your code to reach the functionality I need I would have to call the build within the loop, like this:

 spmd
   x = codistributed.zeros( 100 ); % 100x100 array spread across the workers
   lp = getLocalPart( x ); % The underlying "local part"
   rr = globalIndices( x, 1 ); % Which rows in the global array do we have?
   cc = globalIndices( x, 2 ); % Which columns?
   for ii = 1:length( rr )
     for jj = 1:length( cc )
       % perform some calculation on the local part
       lp(ii, jj) = labindex * (rr(ii) + cc(jj));
       % We've modified the local part of "x", we need to put it back together
       x = codistributed.build( lp, getCodistributor( x ) );
     end
   end
 end

but that throws an error

 communication mismatch error was encountered:
    The other lab became idle during receiving from lab 3 (tag: 32442)

    Error stack:
    gop>gop/receiveWithErrorHandling at 94
    gop.m at 79
    isreplicated.m at 23
    hBuildFromLocalPartImpl>buildWithCompleteDist at 46
    hBuildFromLocalPartImpl.m at 17
    build.m at 107
    smp_soma_all_to_all_adaptive>(spmd body) at 117
....

Placing it anywhere else then you suggested always results into this error.

Until calling the xd = gather( x ); I cannot see the entire current data, so I cannot use this approach... I guess I'll have to try the labSend... will let you know how it ended up.

To Bruno:
Your InplaceArray library is my next move. I hope it is not going to copy the object behind pointers :)

Thanks guys so far, I'll let you know how it goes :)

"ibisek " <cervenka@fai.utb.cz> writes:

> in your code to reach the functionality I need I would have to call the build
> within the loop, like this:
>
> spmd
> x = codistributed.zeros( 100 ); % 100x100 array spread across the workers
> lp = getLocalPart( x ); % The underlying "local part"
> rr = globalIndices( x, 1 ); % Which rows in the global array do we have?
> cc = globalIndices( x, 2 ); % Which columns?
> for ii = 1:length( rr )
> for jj = 1:length( cc )
> % perform some calculation on the local part
> lp(ii, jj) = labindex * (rr(ii) + cc(jj));
> % We've modified the local part of "x", we need to put it back together
> x = codistributed.build( lp, getCodistributor( x ) );
> end
> end
> end
>
> but that throws an error
>
> communication mismatch error was encountered:
> The other lab became idle during receiving from lab 3 (tag: 32442)
>
> Error stack:
> gop>gop/receiveWithErrorHandling at 94
> gop.m at 79
> isreplicated.m at 23
> hBuildFromLocalPartImpl>buildWithCompleteDist at 46
> hBuildFromLocalPartImpl.m at 17
> build.m at 107
> smp_soma_all_to_all_adaptive>(spmd body) at 117
> ....
>
> Placing it anywhere else then you suggested always results into this error.

That's because the call "codistributed.build" is a collective call (it involves
communication) - all workers must ensure that they call that function at the
same time. Inside the loop as you have it, sometimes some workers will call it
and others will not (although you might get lucky if each worker happens to have
the same number of columns of x).

> Until calling the xd = gather( x ); I cannot see the entire current data, so I
> cannot use this approach... I guess I'll have to try the labSend... will let
> you know how it ended up.

Using the low-level message-passing functions such as labSend and labReceive can
be tricky, it might help if you can show the communication pattern you need
between calculations. Is it the case that any worker might need a value from any
other worker, or is there more structure than that?

Cheers,

Edric.

Edric M Ellis wrote in message ...

> Is it the case that any worker might need a value from any
> other worker, or is there more structure than that?

Yes, it is exactly that case. I have a matrix in which rows are 'individuals' (of an evolutionary algorithms) and columns are properties of that individuals. I migrate these individuals (rows) to ALL others (rows) (tha's why I need to have also the rest of the rows accessible) and then to modify a 'working'/current row (to store a modified individual).

Therefore, the functionality I am seeking is something like let's say this in Java:

private volatile float[][] myArray;

public synchronized flloat[] getIndividual(int row) {
     return myArray[row];
}

public synchronized setRow(int rowNumber, float[] row) {
    myArray[rowNumber] = row;
}

Frankly, here I would not be afraid to access the myArray array directly when I take care of not writing into the same row from different threads at the same time. Or more precisely to separate the array by rows so threads can write only into 'their' rows but can read from all, while the data will reflect the actual state of data manipulated by all threads together.

In both, parfor and spmd, I cannot access the fresh data from the other workers. From my point of view spmd does exactly the same as parfor, you just see a little bit more what is happening behind the scenes.

Therefore I think the InplaceArray might help - if Matlab will copy just the pointers, not the data behind. On the other hand, involving a library which is not in the standard Matlab installation might bring some more inconvenience (for users who will want to use the algorithm).

And I am a little worried about the communication overhead the labSend may cause....

Will try both and will report it here soon.

All right, here I am again. I took the InplaceArray trail first. Having this code:

% matlabpool open


len = 10;
A = zeros(len); %just a simple multiplication table

for i=1:size(A,1)
    for j=1:size(A,2)
        A(i,j)=i*j;
    end
end

%create the pointer on the array. further we'll use only the pointer
p = inplacearray(A);

variant = 2;

switch variant
    case 1
        parfor i=1:len
            p(i,2)=i;
            p(i,3)=p(10,10);
        end

    case 2

        %assign array rows to workers:
        len = length(A);
        numWorkers = matlabpool('size');
        indexes = zeros(numWorkers, 2); %from-to
        tasksPerWorker = round(len/numWorkers);

        for i=1:numWorkers
            indexes(i,1) = (i-1)*tasksPerWorker + 1;
            indexes(i,2) = indexes(i,1) + tasksPerWorker - 1;

            if indexes(i,2) > len
                indexes(i,2) = len;
            end
        end

        %do the parallel stuff:
        spmd
            myIndexes = indexes(labindex,:);

            for i=myIndexes(1):myIndexes(2)
                disp(i);
                %p(i,3)=p(10,10);
            end

        end
end

pp = p(:,:);
releaseinplace(p)

pp

% matlabpool close

In variant 1 I get the 'variable p is indexed in different ways' error already in the editor (line with p(i,3)=p(10,10);). Launching it results in

??? Error: The variable p in a parfor cannot be classified.
 See Parallel for Loops in MATLAB, "Overview".

which is quite understandable. And it means, "the thing" beyond parfor attemps to slice the data even if the 'p' variable is a pointer.

In the variant 2 the run results in immediate

------------------------------------------------------------------------
       Segmentation violation detected at Wed Oct 07 14:59:53 2009
------------------------------------------------------------------------
....

even though, THERE IS NO 'p' VARIABLE INVOLVED at all.

I think my desperate attempts to make it work are going to meet my resignation thoughts very soon :(

Somebody has to be facing this problem before. Where are those guys!? Or perhaps I only have a bad google expression....?

Well, I still have the labSend to explore....

"ibisek " <cervenka@fai.utb.cz> writes:

> Edric M Ellis wrote in message ...
>
>> Is it the case that any worker might need a value from any
>> other worker, or is there more structure than that?
>
> Yes, it is exactly that case. I have a matrix in which rows are 'individuals'
> (of an evolutionary algorithms) and columns are properties of that
> individuals. I migrate these individuals (rows) to ALL others (rows) (tha's
> why I need to have also the rest of the rows accessible) and then to modify a
> 'working'/current row (to store a modified individual).

Ok, it seems then that (co)distributed arrays might not be your best bet. You
might be better off using normal arrays, and synchronising them using
communication (by the way, MATLAB workers are all separate processes, so you
cannot share memory between them directly without using special shared memory
segments. And even then, that only works for the special case where the workers
are on the same machine - in general, they might not be). Here's one approach
(using "gcat", which contains calls to labSend and labReceive):

% starting point calculated at the client
X = zeros( 100 );
spmd
    % Because we're using communication, it's important that each worker does
    % the same number of iterations
    for iter = 1:100
        % Lab 1 calculates all the rows to update and broadcasts that to all workers
        rowsToUpdate = labBroadcast( 1, randperm( 100 ) );
        rowsToUpdate = rowsToUpdate( 1:numlabs );
        
        % Each lab selects a given row to update
        rowToUpdate = rowsToUpdate( labindex );
        
        % Calculation: each lab updates a distinct row.
        X( rowToUpdate, iter ) = labindex;

        % synchronise X - first use gcat to transfer updates to each
        % worker. Effectively, each worker sends their updated row to each
        % other worker.
        updates = gcat( X( rowToUpdate, : ), 1 );
        
        % Then apply the updates to X
        X( rowsToUpdate, : ) = updates;
    end
end

% X is a Composite, but each element is identical, so we just need to fetch the
% value from worker 1.
X = X{1};

Cheers,

Edric.

Hi Edric,

If I understand it well, in the loop you posted the workers run synchronously and the labBroadcast command is blocking? Do the others wait until worker #1 sends the message?

Ibisek

"ibisek " <cervenka@fai.utb.cz> writes:

> If I understand it well, in the loop you posted the workers run synchronously
> and the labBroadcast command is blocking? Do the others wait until worker #1
> sends the message?

Yes, the labBroadcast command is effectively blocking - all workers must make
the same labBroadcast call together. (Actually, we don't guarantee whether the
"root" worker sending the data will proceed until the others have received the
data - this depends on the relative size of the data compared to the buffering
in the transport layer, but it's best to assume that the call may be blocking).

Could I ask a little more about why you need the shared access to the array? How
large is the dataset that you'd like to share? How does that compare to the
amount of memory in your machine? How long does each iteration take to run?

Cheers,

Edric.

Allright,

so I managed to make it with the labSend commands. It seems to show quite significant communication overhead when the worker's jobs are short, but we'll see how it performs on a larger jobs....

At this moment it works, as you stated we need to have the same number of jobs for every worker ( mod(numJobs, numLabs)==0). Otherwise it throws an error. I fixed this problem by extending the number of jobs so all workers have the same.

> Could I ask a little more about why you need the shared access to the array? How
> large is the dataset that you'd like to share? How does that compare to the
> amount of memory in your machine? How long does each iteration take to run?

I was developing a parallel version of an evolutionary algorithm. They work with a population of individuals, which are stored in a matrix of (m, n), where 'm' is number of individuals and 'n' length of its 'genome'. In other words, the rows represent individuals and columns are properties of the individuals.

And individual represents a combination of parameters to some model = a possible solution. To improve these solutions we need to somehow handle with these vectors. In our algorithm we migrate one individual to another one. It is something like a wolf is running towards a forest and is looking for some food. A step along this path represents coordinates of its position in the forest. And the amount of food find tells about the quality of that particular solution/position.

But back to the topic. One of the strategies is All-To-One, where there is a best position (individual) and all the other migrate towards that one. This can be easily parallelized by the 'parfor' as the runs do not depend on the other ones. So the matrix is sliced by rows (individuals) and if there are better positions then at the initial moment of their migration, they are stored back to the slice and the entire array of individuals is then de-sliced just after the parfor-end block.

However, in another strategy - All-To-All-Adaptive - you pick individuals one by one and migrate them towards ALL others. You pick a first one and migrate it toward the second one (searching for a better 'source of food'). Once you searched the trail along this line, you go back to the best location/position you found, pick individual #3 and do a migration again. After you migrated to all of the individuals in the population, you STORE the best found location/position/solution back to the SHARED matrix of individuals in a row. And this is the shared array I was looking for.

What we essentially do, is we work upon a population of individuals (the rows in the matrix). The idea is to split this matrix among the Matlab's workers (let's say we have 4 workers, so every worker has a quarter of the rows; the exact fraction is not important - the worker has a list of rows which can modify, never touching other worker's rows (for writing)) and these workers do the job described above. Writing in the rows is restricted only on the 'worker's own' rows, but it can read the current state from the other rows and thus migrate its own individuals towards the others' individuals, which concurrently may be moved into another position.

Does it make any sense? :)

The only collision here may appear in situation when one worker is writing a row with a newly find position while another attempts to read from the same row at the same moment. Result of the read depends on the implementation of Matlab matrices - how they handle concurrent access to variables.

I think adding a functionality of concurrently accessible variables into Parallel Computing Toolkit would ease many people's lives :) On the other hand, it could also be a serious source of problems....

Edric, hereby I would like to thank you very very much for guiding me through this problem and giving me a helping hand. I owe you more than one beer :)

Ibisek

Did it make any sense at all? :)

"ibisek " <cervenka@fai.utb.cz> writes:

>> Could I ask a little more about why you need the shared access to the array? How
>> large is the dataset that you'd like to share? How does that compare to the
>> amount of memory in your machine? How long does each iteration take to run?
>
> I was developing a parallel version of an evolutionary algorithm. They work
>with a population of individuals, which are stored in a matrix of (m, n), where
>'m' is number of individuals and 'n' length of its 'genome'. In other words,
>the rows represent individuals and columns are properties of the individuals.
> [...]

Thanks for taking the time to explain your application. I think I understand the
access patterns you're after now. Unfortunately, our current (co)distributed
arrays are designed to be used for linear algebra, primarily in the case where
the data is too large to fit into the RAM of a single machine. This results in
communication patterns that fit well with the "message passing" approach,
i.e. where both sender and receiver are actively involved, and both know what
data is being exchanged.

In your application, one interpretation is that you want "one-sided"
communication, where each worker can either request data from another worker
without the other worker's involvement, or else each worker can push out updates
to their data to the other workers. Either way, this is not a communication
pattern that we currently support (not least because the underlying technology -
MPI - does not have great support for this; also, there are the synchronization
issues to be taken care of).

I think the only way to address the "random access" nature of your application
is as I described earlier, and to have the workers actively and co-operatively
ensuring that all other workers are up-to-date after each round of calculation.

Cheers,

Edric.

Exactly, the synchronization plays here the main role.

The truth is, I haven't been considering run on multiple machines. Running our algorithm on a single multi-core computer is good enough for us (at least for this moment:). Keeping such a distributed and shared matrix synchronized on multiple machines might become evil, especially when the stuff is automatically done by the logic behind the scene :)

And again, Edric, thank you for the support you've given me, I really really appreciate the time you've spend ! :)

Ibisek

You may want to checkout my Matlab file-exchange submission "sharedmatrix" #28572. It allows a Matlab matrix to exist in shared memory, provided you are using some flavor of Unix. One could then attach the shared matrix in a body of a parfor or spmd, ie,

shmkey=12345;
sharedmatrix('clone',shmkey,X);
clear X;
spmd(8)
   X=sharedmatrix('attach',shmkey);
   % do something with X
   X=sharedmatrix('detach',shmkey);
end
sharedmatrix('free',shmkey);

Cheers,

Josh

http://www.mathworks.com/matlabcentral/fileexchange/28572-sharedmatrix

Hi, I see that you are interested about Algorithms and programing so maybe I can help you? I find one amazing book about it "Advances in Evolutionary Algorithms". This is the link where you can find it and download for free: http://www.intechopen.com/books/show/title/advances_in_evolutionary_algorithms Aim of the book is to present recent improvements, innovative ideas and concepts in a part of a huge EA field.