DataMuseum.dk

top - metrics - download

    Length: 16128 (0x3f00)
    Types: TextFile
    Names: »tbatch«

└─⟦00964e8f7⟧ Bits:30007478 RC8000 Dump tape fra HCØ.
    └─⟦b2ec5d50f⟧ 
        └─⟦0b817e319⟧ »ctramos« 
            └─⟦this⟧ 

TextFile

batch queue 82.01.06
Anders Lindgård
:1: batch queue: trim variables and fields
algol list.on;
integer qdescsize,qdes,b_max_jobs,b_curjob,
        b_child_first,b_child_last,
        b_max_time,
        b_max_time_day,  b_max_size_day,
        b_max_time_night,b_max_size_night,
        b_day_to_night,  b_night_to_day,
        b_upd_time,b_interval,b_time_slice,b_std_time,
        b_max_children,b_cur_children,b_job_number,
        b_run_children,b_stop_begin,b_stop_end,
        swop_segm,
        b_max_prio_day;
real    b_time_fac,b_job_fac;
integer field q_evening,q_link_next,q_link_last,
              q_job_no,q_job_state,q_job_seq,q_max_time,
              q_start_time,c_start_time,c_removed_time,
              q_remove_job_file,q_lref;
long array field q_job_name,q_printer,q_wrk_o,
                 d_job_name;
integer field q0_max_time_day,q0_max_size_day,
              q0_max_time_night,q0_max_size_night,
              q0_day_to_night,q0_night_to_day,
              q0_upd_time;
long array qname,swop,swopdoc(1:3);
algol list.off;
:2: batch queue: initialization
algol list.on;
d_job_name:=4;
i:=qdescsize:=condescsize;
i:=q_link_next:=i+2;
i:=q_link_last:=i+2;
i:=q_jobno:=i+2;
i:=q_evening:=i+2;
i:=q_job_state:=i+2;
i:=q_job_seq:=i+2;
i:=q_max_time:=i+2;
i:=q_start_time:=i+2;
i:=q_remove_job_file:=i+2;
i:=q_lref:=i+2;
i:=q_job_name:=i;
i:=q_printer:=i+8;
i:=q_wrk_o:=i+8;
i:=c_start_time:=i+2+8;
i:=c_removed_time:=i+2;
i:=qdescsize:=i;
<*global information*>
i:=q0_max_time_day:=q_link_last+2;
i:=q0_max_size_day:=i+2;
i:=q0_max_time_night:=i+2;
i:=q0_max_size_night:=i+2;
i:=q0_night_to_day:=i+2;
i:=q0_day_to_night:=i+2;
i:=q0_upd_time:=i+2;
qname(1):=long <:bqueu:> add 'e';
qname(2):=0;
b_max_time_day:=20*60; <*sec*>
b_max_time_night:=8*60*60;
b_max_size_day:=512*coreblocksize;
b_cur_job:=0;
b_maxjobs:=250;
b_max_children:=1;
b_std_time:=5*60;
b_time_slice:=30;
b_interval:=5;
swop(1):=long <:bswop:>; swop(2):=0;
for i:=1,2 do swopdoc(i):=owndisc(i);
b_time_fac:=0.5;
b_job_fac :=1.0;
algol copy.batchtrim;
readifp(<:btimeslice:>,b_time_slice,b_time_slice);
b_max_prio_day:=2*b_max_time_day;
readifp(<:bprio:>,b_max_prio_day,b_max_prio_day);
readrfp(<:btimefac:>,btimefac,btimefac);
readrfp(<:bjobfac:>,bjobfac,bjobfac);
readifp(<:bchildren:>,bmaxchildren,bmaxchildren);
readifp(<:bjobs:>,bmaxjobs,bmaxjobs);
readifp(<:bstdtime:>,bstdtime,bstdtime);
readifp(<:time:>,b_max_time_day,b_max_time_day);
laf:=2;
swopsegm:=bmaxchildren*bmaxsizenight//512+1;
b_curchildren:=0;
qdes:=maxchildren+1;
b_job_number:=0;
if readifp(<:bsize:>,bmaxsizeday,bmaxsizeday) then
  bmaxsizeday:=bmaxsizeday*1024; <*halfwords*>
<*+2*>
if testop(7) then write(out,"nl",1,<:qdes, qdeszise:>,
    qdes,qdescsize);
<*-2*>
algol list.off;
:3: batch que: claiming basic
maxcoru:=maxcoru+4;
maxsemch:=maxsemch+2;
maxop:=maxop+qdes+b_max_jobs+4;
maxnettoop:=maxnettoop+12*qdes+6*(bmaxjobs+4);
maxsem:=maxsem+qdes+2;
maxmessext:=maxmessext+1;
:4: batch queue global variables and procedures
algol list.on;
integer array q(1:qdes*qdescsize//2);
integer array b_segm_table(1:b_maxchildren,1:2);
integer array job_table(1:b_maxjobs,1:15);
<* 1: segment number
   2: state (0: free, 1: enrolled, 2: running, 3: removed)
   3: unique job number
   4: current job priority
   5: run time
   6: rest time
   7: childno
   8: termpda
   9: jobprio
  10: projno
  11: cpu time
  12: sequence number
  13: holdjob (=evening job)
  14: link to next job this project
  15: link to last job this project
*>
zone wrp(128,1,stderror),qz,q_in(128,1,noerror);
real timebase;
boolean cmess,qmess,bmess,jobtimermess,newqueue,day;
integer qmessline,bmessline,qsem,wrpsem;

algol list.on copy.treadsub;

procedure init_batch_queue;
begin
if -,newqueue then
begin
integer array field bref;
integer segm;
<*+2*>
  if testop(7) then
  begin
   disable write(out,"nl",1,<:b queue :>,qname);
   outendcur(0);
  end;
  if testop(2) then disable
  begin
    write(out,"nl",1,<:segm 0 (before) :>,qname);
    outendcur(0);
  end;
<*-2*>
  get_job_segm_0(false);
  for segm:=1 step 1 until b_max_jobs do
  begin
   if get_job_segm(q,segm,false) then
   begin
     <*old job found*>
     if q.qjobstate=staterunning then
     begin
       disable write(out,"nl",1,true,12,q.qjobname,q.qjobno,<: lost:>);
       q.qjobstate:=q.qjobno:=0;
       get_jobsegm(q,segm,true);
     end else
     begin
     b_cur_job:=b_cur_job+1;
<*+2*>
     if testop(7) then
     disable begin
       write(out,"nl",1,<:job :>,b_curjob,segm);
      outendcur(0);
     end;
<*-2*>
     jobtable(segm,1):=segm;
     jobtable(segm,2):=q.q_job_state;
     jobtable(segm,3):=q.q_job_no;
     jobtable(segm,4):=q.q_maxtime;
     jobtable(segm,5):=q.qmaxtime;
     jobtable(segm,6):=q.qmaxtime;
     jobtable(segm,7):=0;
     jobtable(segm,8):=q.contermpda;
     jobtable(segm,9):=q.qmaxtime;
     jobtable(segm,10):=q.conprojno shift (-8) extract 16;
     jobtable(segm,11):=0;
     jobtable(segm,12):=0;
     jobtable(segm,13):=q.q_evening;
     jobtable(segm,14):=q.q_link_next;
     jobtable(segm,15):=q.q_link_last;
     if bjobnumber<q.fi.q_job_no then b_jobnumber:=
        q.fi.q_job_no;
      waitch(bmessline,bref,free,0);
      d.bref(1):=2;
      d.bref(2):=segm;
      signalch(bmessline,bref,bmess);
   end initjob;
   end old job;
  end scan segments;
end -,newqueue;
disable write(out,"nl",2,<:batch included :>,if newqueue then
     <:new:> else <:old:>,<: queue :>);
disable writetime(out,b_upd_time);;
disable write(out,<< dddddd>,
          "nl",1,<:current max time:>,bmaxtime,
            if day then <: day:> else <: night:>,
          "nl",1,<:max time day    :>,b_max_time_day//60,<: min:>,
          b_max_time_day mod 60,<: sec:>,
          "nl",1,<:max size day    :>,b_max_size_day//1024,<: k:>,
          "nl",1,<:max time night  :>,b_max_time_night//60//60,<: h:>,
          "nl",1,<:max size night  :>,b_max_size_night//1024,<: k:>,
          "nl",1,<:day to night    :>,<<_____dd.dd>,b_day_to_night//100,
          "nl",1,<:night to day    :>,b_night_to_day//100,
          "nl",1,<:parallel jobs   :>,<< dddddd>,b_max_children,
          "nl",1,<:jobs in queue   :>,b_cur_job,
          "nl",1,<:max job number  :>,b_job_number);
outendcur(10);
end initbatch;

procedure batch_queue;
begin
integer i,j,res,cbn,op,mode,word,jobno,jobstate;
long array field nf;
integer array field qref,qcur,concur,bref,ct;
long array name,text(1:3);
zone pz(17,1,noerror);

stackclaim(800);
fi:=0;
<*+2*>
if testop(7) then writelog(<:batch queue started:>,0,<::>);
<*-2*>
repeat
  waitch(qmessline,qref,qmess,0);
  concur:=d.qref(2);
  for i:=1,2 do name(i):=condesc.concur.condesterm(i);
  open(pz,8,name,1 shift 9);
  open(qin,8,name,1 shift 9);
<*+2*>
  if testop(7) then
  disable begin
    write(pz,"nl",1,<:batch called :>,concur,
     case d.qref(1) of (<:submit:>,<:queue:>,<:what:>,
          <:search:>,<:kill:>,<:release:>,<:setprio:>,<::>));
    if d.qref(1)=1 then write(pz,d.qref.djobname,qref,djobname) else
    write(pz,d.qref(3));
  end test;
<*-2*>
  case d.qref(1) of
  begin
    begin
      <*submit*>
      qcur:=0;
      j:=condescsize//2;
      for i:=2 step 1 until j do
        q.qcur(i):=condesc.concur(i);
      j:=j+1;
      for i:=qdescsize//2 step -1 until j do
          q.qcur(i):=0;
      res:=0;
      if d.qref.d_job_name(1)<>0 then
      begin
<*+2*>
        if testop(7) then write(pz,"nl",1,<:b: search user cat :>,
           d.qref.d_job_name);
<*-2*>
        if readusercat(d.qref.d_jobname,q.qcur,testop(7),pz) then
        begin
          stdclaim(q.qcur);
          stdbs(q.qcur,pz);
        end else
        begin
          res:=4;
          writeerror(pz,errornameunknown,q.qcur);
        end;
      end jobname;
      if res=0 then read_submit(qin,pz,q.qcur,false);
    end submit;
    begin
    integer t;
      <*queue*>
      t:=condesc.concur.contermpda;
      list_job_table(pz,if t=sysconpda then 0 else t);
    end queue;
    begin
    integer i,jte,rest;
    boolean first;
      <*what*>
       first:=true;
       for i:=1 step 1 until maxchildren do
       begin
        ct:=(i-1)*ctsize;
        if childtable.ct.ct_batch>0 then
        begin
          if first then
          begin
             write(pz,"nl"1,if day then <:day :> else <:night :>);
             writecurtime(pz);
             write(pz,"nl",1,<:user________:>,<:state____:>,
               <:jobno__:>,<:time left:>);
            first:=false;
          end;
          jte:=childtable.ct.ct_segm_queue;
          rest:=jobtable(jte,6);
          write(pz,"nl",1,true,12,childtable.ct.ct_jobname,
            true,9,case childtable.ct.ctstate of
             (<:created:>,<:running:>,<:stopped:>,<:breaked:>,
              <:swopped:>),<< dddd>,childtable.ct.ctjobno);
          if rest>0 then write(pz,<< dd >,rest//3600,rest mod 3600 //60,
               rest mod 60);
         end batch job;
       end childtable scan;
    end what;
    begin
      <*search*>
<*+2*>
       if testop(7) then disable writebdes(q.qcur,pz);
<*-2*>
       i:=0;
       jobno:=d.qref(3);
       repeat i:=i+1;
       until i=bmaxjobs or jobno=jobtable(i,3);
       if jobno=jobtable(i,3) then
       begin
         write(pz,"nl",1,<:job number:>,jobno,<: :>,
           case jobtable(jobno,2)+1 of (
          <:impossible:>,<:enrolled:>,<:running:>,<:removed:>));
        end else writeerror(pz,errorjobnumbernotfound,q.qcur);
    end search;
    begin
      <*kill*>
      i:=0;
      jobno:=d.qref(3);
      repeat i:=i+1;
      until i=bmaxjobs or jobno=jobtable(i,3);
      if jobno=jobtable(i,3) then
      begin
        jobstate:=jobtable(i,2);
        b_cur_job:=b_cur_job-1;
        if jobstate=2 then
        begin
          waitch(bmessline,bref,free,0);
          d.bref(1):=1; <*clock*>
          d.bref(2):=-6; <* kill running job*>
          d.bref(3):=(jobtable(i,7)-1)*ctsize;
          signalch(bmessline,bref,cmess);
        end job running else
         for j:=2 step 1 until 15 do jobtable(i,j):=0;
      qcur:=0;
      q.qcur.qjobno:=0;
      get_job_segm(q.qcur,i,true);
      end else writeerror(pz,errorjobnumbernotfound,q.qcur);
    end kill;
    begin
      <*release*>
       i:=0;
       jobno:=d.qref(3);
       repeat i:=i+1;
       until i=bmaxjobs or jobno=jobtable(i,3);
       if jobno=jobtable(i,3) then
       begin
         <*job no found*>
         jobtable(i,13):=0;
       end job found else
       writeerror(pz,errorjobnumbernotfound,q.qcur);
    end release;
    begin
      <*set prio*>
       i:=0;
       jobno:=d.qref(3);
       repeat i:=i+1;
       until i=bmaxjobs or jobno=jobtable(i,3);
       if jobno=jobtable(i,3) then
       begin
          jobtable(i,4):=0;
       end else writeerror(pz,errorjobnumbernotfound,q.qcur);
    end setprio;
  end case;
<*+2*>
  if testop(7) then writelog(<:batch end:>,0,<::>);
  if testop(7) then
  begin
    write(pz,"nl",1,<:batch :>,d.qref(1),
     d.qref(2));
  if d.qref(1)=1 then write(pz,<: :>,d.qref.d_jobname) else
    write(pz,d.qref(3));
  end test;
<*-2*>
  close(pz,true);
  close(q_in,true);
signalch(qmessline,qref,free);
until false;
end batch queue;

procedure b_timer;
begin
integer array field bref;
real r;
stackclaim(250);
repeat
  waitch(bmessline,bref,free,0);
  delay(b_time_slice);
  systime(5,0,r);
  d.bref(1):=1;
  d.bref(2):=r;
  signalch(bmessline,bref,cmess);
until false;
end b_timer;

procedure run_job_timer;
begin
integer array field bref;
  stackclaim(150);
  repeat
  waitch(bmessline,bref,jobtimermess,0);
  d.bref(1):=2;
  d.bref(2):=-1;
  delay(b_time_slice);
  signalch(bmessline,bref,bmess);
  until false;
end runjobtimer;
algol list.on copy.trunbatch;

procedure write_b_des(d,z);
integer array d;
zone z;
begin
  writeentry(d,z,true);
  write(z,"nl",1,<:job number   :>,d.qjobno,
          "nl",1,<:job state    :>,d.qjobstate,
          "nl",1,<:job seq      :>,d.qjobseq,
          "nl",1,<:max time     :>,d.qmaxtime,
          "nl",1,<:start wanted :>,d.qstarttime,
          "nl",1,<:job file name:>,d.qjobname,
          "nl",1,<:printer      :>,d.qprinter,
          "nl",1,<:started      :>,d.cstarttime,
          "nl",1,<:removed      :>,d.cremovedtime);
end writebdes;


algol list.off;
:5: batch queue: program
algol list.on;
setbasestd;
tail(1):=swopsegm;
laf:=2;
for i:=1,2 do tail.laf(i):=swopdoc(i);
for i:=6 step 1 until 10 do tail(i):=0;
i:=createentry(swop,tail);
if i=3 then
begin
  removeentry(swop);
  i:=createentry(swop,tail);
end;
if i>0 then alarm(<:***swop area :>,swop,i);
permentry(swop,3);
resetbase;
begin
integer cs,cb,blocks;
  cs:=coreblocksize*512;
  blocks:=(bmaxsizeday+cs-2)//cs;
  if blocks<=noofcoreblocks then
  begin
   cb:=0;
   for cb:=1 step 1 until blocks do coretable(cb,2):=ownpda;
   bchildfirst:=coretable(1,1);
   bchildlast :=coretable(blocks+1,1)-2;
  end else
  begin
    write(out,"nl",1,<:batch inactive, no free core :>,
    bmaxsizeday,<: needed:>);
  bmaxchildren:=0;
  end;
end local;
qmess:=false add (1 shift 5);
bmess:=false add (1 shift 6);
cmess:=false add (1 shift 7);
jobtimermess:=false add (1 shift 8);
wrpsem:=nextsem;
signal(wrpsem);
qsem:=nextsem;
signal(qsem);
qmessline:=nextsemch;
for i:=1 step 1 until qdes do
begin
  j:=nextop(12);
  signalch(qmessline,j,free);
end;
systime(5,0,timebase);
timebase:=timebase/100;
day:=(timebase>b_night_to_day and timebase<b_day_to_night);
b_max_time:=if day then b_max_time_day else b_max_time_night;
bmessline:=nextsemch;
for i:=1 step 1 until bmaxchildren do
begin
  bsegmtable(i,1):=0;
  bsegmtable(i,2):=(i-1)*bmaxsizenight//512;
end;
for i:=1 step 1 until b_maxjobs+4 do
begin
  j:=nextop(6);
  signalch(bmessline,j,free);
end;
for i:=1 step 1 until b_max_jobs do
for j:=1 step 1 until 10 do
job_table(i,j):=0;
setbaseusercat;
readbfp(<:newq:>,newqueue,false);
i:=lookupentry(qname,tail);
if i>0 or tail(1)<=b_max_jobs  or newqueue then
begin
  newqueue:=true;
  tail(1):=b_max_jobs+1;
  laf:=2;
  for i:=1,2 do tail.laf(i):=own_disc(i);
  for i:=7 step 1 until 10 do tail(i):=0;
  removeentry(qname);
  i:=createentry(qname,tail);
  if i>0 then alarm(<:Queue no resources :>,qname,i);
end;
permentry(qname,3);
open(qz,4,qname,1 shift 9);
if newqueue then
begin
  outrec6(qz,512);
  fi:=0;
  for i:=1 step 1 until 256 do qz.fi(i):=0;
  qz.fi.q0_max_time_day:=b_max_time_day;
  qz.fi.q0_max_size_day:=b_max_size_day;
  qz.fi.q0_max_time_night:=b_max_time_night;
  qz.fi.q0_max_size_night:=b_max_size_night;
  qz.fi.q0_night_to_day:=b_night_to_day;
  qz.fi.q0_day_to_night:=b_day_to_night;
  qz.fi.q0_upd_time:=b_upd_time:=systime(7,0,0.0);
  outrec6(qz,512);
  for i:=1 step 1 until 256 do qz.fi(i):=0;
  for i:=3 step 1 until b_max_jobs+1 do outrec6(qz,512);
  setposition(qz,0,0);
end newqueue;
resetbase;
for i:=1 step 1 until qdes do
begin
  fi:=(i-1)*qdescsize;
  for j:=2 step 1 until qdescsize//2 do q.fi(j):=0;
  q.fi.conref:=fi;
  q.fi.conaccess:=nextsem;
  signal(q.fi.conaccess);
end;
j:=nextcoru(4,300,true);
<*+2*>
if testop(7) then write(out,"nl",1,<:batch queue corutine :>,j);
<*-2*>
newactivity(j,j,batch_queue);
j:=nextcoru(5,50,true);
<*+2*>
if testop(7) then write(out,"nl",1,<:run batch corutine :>,j);
<*-2*>
newactivity(j,j,run_batch);
j:=nextcoru(6,25,true);
<*+2*>
if testop(7) then write(out,"nl",1,<:b timer corutine :>,j);
<*-2*>
newactivity(j,j,b_timer);
j:=nextcoru(7,25,true);
<*+2*>
if testop(7) then write(out,"nl",1,<:run job timer corutine:>,j);
<*-2*>
newactivity(j,j,run_job_timer);
algol list.off;
▶EOF◀