Each PARFOR Worker Writes to the Same File

I understand that having each worker write to a single file is a no-no. Perhaps as expected, when I run this code, it shows some corrupted values in the final output file; about 10% fails.

I have no interest in the output being in a deterministic order. The workers are spread across multiple Linux machines. The amount of time to complete a run is long compared to the time to write a single line of output.

Can someone recommend an alternative?

% Run a parametric study

var1 = (-60:0.5:60)';
var2 = (-110:0.5:110)';
var3 = (3.5:0.5:18.5)';

% Remove zero entries since their usage prohibited
var1(var1 == 0) = [];
var2(var2 == 0) = [];
var3(var3 == 0) = [];

NS = length(var1)*length(var2)*length(var3); % Number of runs

% Set up the design matrix, desMat
desMat = {var1,var2,var3};
[desMat{:}]=ndgrid(desMat{:});
n=length(desMat);
desMat = reshape(cat(n+1,desMat{:}),[],n);

if exist('./Results.csv', 'file')==2
delete('./Results.csv');
end

parfor kk = 1:NS

var1a = desMat(kk,1); var2a = desMat(kk,2); var3a = desMat(kk,3);
[out1 out2 out3] = Function_Pd(var1a,var2a,var3a);

    vec = [var1a var2a var3a out1 out2 out3];
    fileID = fopen('Results.csv','a');
    fprintf(fileID,'%f %f %f %f %f %f\n',vec);
    fclose(fileID);
    
end

[jessupj]

i don't have a solution. YOu can probably just change

fileID = fopen('Results.csv','a');

to

fileID = fopen( sprintf('Results_%02d.csv',kk),'a');

Unrecognized function or variable 'kk'.

to save things in different files and then assemble them afterward.

Also, you might check out the exchange if you're just using parallelization for speed without requiring any interim communincation: https://www.mathworks.com/matlabcentral/fileexchange/13775-multicore-parallel-processing-on-multiple-cores

[Paul Safier]

@jessupj The individual files would work except there would be over 3 million files and I believe file servers can have issues with that many...

Answer 1

[Bruno Luong]

May be (I didn't test) you could write in binary file at a deterministic place:

fileID = fopen('Results.bin','wb');
parfor ...
    ...
    fseek(fileID, (kk-1)*length(vec)*8, 'bof'); % 8 is byte size of double, assuming vec is double
    fwrite(fileID, vec);
end
fclose(fileID);

[Paul Safier]

Thanks for this suggestion @Bruno Luong, I will give it a shot. A couple of questions. If you write in a binary file do you avoid the issue of workers accessing it simultaneously? Is this because the specific write location can be stipulated? Second, is it straightforward to then read the binary file and convert to ascii?

[Walter Roberson]

Note that in order to fseek() into a file properly, the file has to be at least that long already. You would need to pre-create the file to the desired size.

This differs from POSIX: POSIX says that if you fseek() past the end of the file and then write to the file, that the result is well defined, that zeros will be put into the file in the gap (either directly or through virtual blocks, up to the operating system in question.)

MATLAB practice is that if you fseek() past the end of the file, then you end up at the end of the file.

You might want to use memmapfile()

[Bruno Luong]

@Walter Roberson: good tip

Another potential issue of my suggestion is that fseek/fwrite are not atomic so others workers can run fseek in between. It needs to be somewhat protected by a "non-interuptible" (CriticalSection on Windows). I have no idea if such mechnism exsists in MATLAB Parallel Computing.

[Bruno Luong]

@Paul Safier "Second, is it straightforward to then read the binary file and convert to ascii?"

Yes just read the file by chunks depending on your RAM available then write to ascii file.

But if whatever app that needs those data can read binary file, why not leave it alone. It's luch better than ascii file: smaller, faster, no precision lost.

[Paul Safier]

@Walter Roberson thanks for the advice. I will look into how to use memmapfile. It may also be an option to fill a binary file with garbage before I start the parfor loop, then overwrite it as I go because of the nature of fseek you mention. There's still the option that @Jeff Miller brings to light, namely having each worker write to its own file than I can cat them at the end, however that may take some leg work to get running. Thanks for the suggestion about the binary file, @Bruno Luong.

Answer 2

[Jeff Miller]

Maybe have each worker write to its own output file and then assemble those after? This answer shows how to get the id for each worker.

[Bruno Luong]

The same suggestion has been discussed in the comment below the question

[Jeff Miller]

Oh, sorry, I thought that suggestion was to write one file for each iteration of the parfor loop rather than for each separate worker.

[Paul Safier]

@Jeff Miller I think it was to write each iteration to its own file, which would be too much for the file server (>3 million files). Your suggestion about each worker writing to its own file would be a managable amount of files. I will look into the link you sent. Thanks.

Answer 3

[Raymond Norris]

@Paul Safier since the order of the file doesn't have to be deterministic, use a data queue to write back to the client and have the client write the csv file.

% Run a parametric study
var1 = (-60:0.5:60)';
var2 = (-110:0.5:110)';
var3 = (3.5:0.5:18.5)';

% Remove zero entries since their usage prohibited
var1(var1 == 0) = [];
var2(var2 == 0) = [];
var3(var3 == 0) = [];

NS = length(var1)*length(var2)*length(var3); % Number of runs

% Set up the design matrix, desMat
desMat = {var1,var2,var3};
[desMat{:}]=ndgrid(desMat{:});
n=length(desMat);
desMat = reshape(cat(n+1,desMat{:}),[],n);

if exist('./Results.csv', 'file')==2
    delete('./Results.csv');
end
fileID = fopen('Results.csv','a');
D = parallel.pool.DataQueue;
afterEach(D,@(V)logger(fileID,V))
c = onCleanup(@()fclose(fileID));
    
parfor kk = 1:NS
    var1a = desMat(kk,1); var2a = desMat(kk,2); var3a = desMat(kk,3);
    [out1 out2 out3] = Function_Pd(var1a,var2a,var3a);
    vec = [var1a var2a var3a out1 out2 out3];
    send(D,vec)
end

function logger(fileID,vec)
fprintf(fileID,'%f %f %f %f %f %f\n',vec);
end

[Paul Safier]

Thanks @Raymond Norris, I will try this!

[Paul Safier]

@Raymond Norris. This worked great, thanks!

When everything was complete, I typed clear to clean the workspace. This warning came up. Do you know why?

Warning: The following error was caught while executing 'onCleanup' class destructor:
Error using fclose
Invalid file identifier.  Use fopen to generate a valid file identifier.

Error in para_varTimeStep_Code1>@()fclose(fileID) (line 124)
c = onCleanup(@()fclose(fileID));

Error in onCleanup/delete (line 80)
obj.task(); 

[Raymond Norris]

@Paul Safier somewhere/how, you've already closed your file. I can reproduce your warning here

fileID = fopen('Results.csv','a');
c = onCleanup(@()fclose(fileID));
fprintf(fileID,"%f\n",rand);
fclose(fileID);

>> safier
>> clear
Warning: The following error was caught while executing 'onCleanup' class destructor:
Error using fclose
Invalid file identifier.  Use fopen to generate a valid file identifier.
Error in safier>@()fclose(fileID) (line 2)
c = onCleanup(@()fclose(fileID));
Error in onCleanup/delete (line 23)
            obj.task();            

I wouldn't have gotten the warning if I hadn't called

fclose(fileID);

[Paul Safier]

@Raymond Norris. Thanks.

Answer 4

[Alexander Denman]

One option you might consider is using a database, for example PostgreSQL. Ensuring that concurrent writes don't interfere with each other is one of the core functions of a relational database management system.

To do this, you would install PostgreSQL on a machine that is reachable from all of your worker nodes, then create a table to hold the results. You can store essentially any matlab variable in a postgres "bytea" colum by using typecast(getByteStreamFromArray(someVariable),'int8') to convert the variable to one long stream of 8 bit integers.

Then, when each worker is ready to save its results, it opens a database connection using the 'database' function, uploads the results using 'sqlwrite' or 'datainsert', and then closes the connection.

When you retrieve the results from the database, you r-convert the data to its original form using getArrayFromByteStream(typecast(binaryDataFromDatabase),'uint8')).

[Paul Safier]

@Alexander Denman, thanks for this interesting option. I will keep this in mind. I have adopted the approach suggested by @Raymond Norris above and it works perfectly. Thanks for you input.