function [B E]=pkriging(x,y,observations,x_interp,y_interp,varargin)
%PKRIGING - parallel Kriging
%
%   Usage:
%      [B E]=pkriging(x,y,observations,x_interp,y_interp,varargin);

options=pairoptions(varargin{:});
cluster=getfieldvalue(options,'cluster',generic('np',1));
options=removefield(options,'cluster',0);
name   = ['krig' num2str(feature('GetPid'))];

% =========================================   MARSHALL.m =================================================
disp(['marshalling file ' name '.bin']);
fid=fopen([name '.bin'],'wb');
if fid==-1,
	error(['marshall error message: could not open ' name '.bin file for binary writing']);
end

%First, write MaximumNumberOfDefinitionsEnum to make sure that the Enums are synchronized
WriteData(fid,'enum',MaximumNumberOfDefinitionsEnum(),'data',true,'format','Boolean');

%Write all data
WriteData(fid,'enum',0,'data',x,'format','DoubleMat');
WriteData(fid,'enum',1,'data',y,'format','DoubleMat');
WriteData(fid,'enum',2,'data',observations,'format','DoubleMat');
WriteData(fid,'enum',3,'data',x_interp,'format','DoubleMat');
WriteData(fid,'enum',4,'data',y_interp,'format','DoubleMat');

%Last, write MaximumNumberOfEnum+1 to make sure that the binary file is not corrupt
WriteData(fid,'enum',MaximumNumberOfDefinitionsEnum()+1,'data',true,'format','Boolean');

%Now, write number of options
options.marshall(fid,5);
st=fclose(fid);
if st==-1,
	error(['marshall error message: could not close file ' name '.bin']);
end
% =========================================   MARSHALL.m =================================================

%Launch job on remote cluster
BuildKrigingQueueScript(cluster,name,'',1,0,0); %gather, valgrind, gprof
LaunchQueueJob(cluster,name,name,{[name '.bin'] [name '.queue']});

%Call waitonlock
md=model; md.cluster=cluster; md.settings.waitonlock=Inf; md.private.runtimename=name;md.miscellaneous.name=name;
waitonlock(md);

%Download
Download(cluster,name,{[name '.outbin']});
structure=parseresultsfromdisk([name '.outbin'],0);
delete([name '.outlog']);
delete([name '.errlog']);
delete([name '.outbin']);
delete([name '.bin']);
if ~ispc(),
	delete([name '.tar.gz']);
end

%Process results
B=structure.(EnumToString(0));
B=reshape(B,size(x_interp,2),size(x_interp,1))';
E=structure.(EnumToString(1));
E=reshape(E,size(x_interp,2),size(x_interp,1))';