unit uoddface_simple;

interface

uses
  Classes, SysUtils,
  Math,
  urosx,
  xmatdefs,
  uauxiliares,
  process,
  uConstantesSimSEE,
  uDataSetGenerico;

const
  // constantes modificables por el usuario
  verboso = 0; // poner a 0 para evitar salidas que no sean de error.
  ntares_por_vez = 10; // cantidad de tareas que intenta lazar cada vez.

  // poner a TRUE para pasarle a la aplicaci�n un idEjecutor �nico.

  pausa_segs = 10;
  // pausa en segundo entre lanzamientos. Para darle tiempo a cada demonio a copiar los datos comunes a su /tmp sin interferirse.

  qstat_cmd__ = 'tasklist';
  crontab_cmd__ = 'schtasks'; // '/usr/bin/crontab';



type

  { TConfigRec }

  TConfigRec = class
    IpFija: string;
    Host: string;
    Puerto: string;
    uri: string;
    usuario_http: string;
    clave_http: string;
    inodo: integer;
    carpeta_tmp: string;
    maxnucleos: integer;
    maxtareas: integer;
    maxhoras: integer;
    constructor CreateFromArchi( archi: string );
    procedure WriteToArchi( archi: string );
  end;


  { TCola }
  TCola = class
    nombre: string;
    nnucleos: integer;
    maxtareas: integer;
    maxhoras: integer;
    constructor Create(nombre: string; nnucleos, maxhoras, maxtareas: integer);
  end;

  { TRBT_OddFace_Simple }

  TRBT_OddFace_Simple = class
    usuario: string;
    home_: string;
    yo: string;
    aplicacion: string;
    orden: string;
    colas: TList;
    ncolas: integer;
    s: ansistring;
    sql: string;
    db: TDBrosxCon;
    ds: TResultadoQuery;

    cfg: TConfigRec;

    constructor Create;

    // ejecuta el comando y retorna la última linea. En mls retorna un stringlist con todas las lineas de salida.
    function exec(apl: string; params: array of string;
      var mls: TStringList; flgWaitOnExit: boolean = true ): string; overload;
    function exec(comando: string; params: array of string;
      flgWaitOnExit: boolean=true): string; overload;

    // Recibe un array de constantes y retorna un TDAofString con las mismas
    function daofs(a: array of string): TDAofString;

    // writeln si vervoso > 1
    procedure wrln(s: string);

    // writeln independientemente del valor de vervoso.
    procedure wrln_err(s: string);


(*
  nexval( $key, $reserva = 1) , retorna un identificador único global
  el parámetro $reserva indica la cantidad de identificadores consecutivos
  reservados (es decir asigandos y que no volveran ser asignados.
 *)
    function nextval(key: string; reserva: integer = 1): integer;


    procedure remover_crontab;
    procedure instalar_crontab;

    // retorna la cantidad de trabajos esperando para ser despachados
    function getRunning: integer;

    // emulador de substr de php
    function substr(const s: string; k_base_0: integer; cnt: integer = -1): string;

    procedure Ayuda;
    procedure lanzar(ntareas: integer; xcola: TCola);
    procedure runLanzarParar(xcola: TCola);
    procedure parar(ntareas: integer);

    // retorna TRUE si la orden es válida.
    function ProcesarOrden: boolean;

    procedure orden_run;
    procedure order_parar_tareas;

    procedure dispatch_cmd;


  private
    procedure orden_lst;
    procedure orden_cancelartrabajos(username, xcola, wk_nidProblema: string);
  end;


implementation

type
  TDAofInteger = array of integer;


function MedirCols(r: string; ColsCap: array of string): TDAOfInteger;
var
  k, i: integer;
  res: TDAOfInteger;
begin
  setlength(res, length(ColsCap));
  for k := 0 to high(ColsCap) do
  begin
    i := pos(ColsCap[k], r);
    if i = 0 then
      raise Exception.Create('ColsCap no encontrada: ' + ColsCap[k]);
    res[k] := i + length(ColsCap[k]);
  end;
  Result := res;
end;

function getColVal(r: string; icols: TDAofInteger; kCol: integer): string;
var
  res: string;
begin
  if kCol = 0 then
    res := copy(r, 1, icols[0] - 1)
  else
    res := copy(r, icols[kCol - 1], icols[kCol] - icols[kCol - 1]);
  Result := trim(res);
end;

{ TConfigRec }

constructor TConfigRec.CreateFromArchi(archi: string);

var
  f: TFileStream;


function rdln: string;
var
  res: string;
  c: char;
begin
  res:= '';
  while (f.Read( c, sizeOf( c ) ) = 1) and ( c <> #10 ) do
     res:= res + c;
  while (length( res ) > 0) and ( res[length( res ) ] = #13 ) do delete( res, length(res), 1 );
  result:= res;
end;

function rp: string;
var
  r: string;
  i: integer;
begin
  r:= rdln;
  i:= pos( '//', r );
  if i > 0 then
    delete( r, i, length( r ) -i+1 );
  result:= trim( r );
end;


begin
  if not FileExists( archi ) then
    WriteToArchi( archi ); // creamos uno vacío.
  f:= TFileStream.Create( archi, fmOpenRead + fmShareDenyWrite );
  IpFija:= rp;
  Host:= rp;
  Puerto:= rp;
  uri:= rp;
  usuario_http:=rp;
  clave_http:= rp;
  inodo:= strToInt( rp );
  carpeta_tmp:= rp;
  maxnucleos:= StrToInt( rp );
  maxtareas:= StrToInt( rp );
  maxhoras:= StrToInt( rp );
  f.Free;
end;



procedure TConfigRec.WriteToArchi(archi: string);

var f: TFileStream;

procedure wrln( s: string );
var
  ts: string;
begin
  ts:= s + #13#10;
  f.WriteBuffer( ts[1], length( ts ) );
end;

begin
  f:= TFileStream.Create( archi, fmCreate  );
  wrln( IpFija+' // IpFija' );
  wrln( Host+' // Host' );
  wrln( Puerto+' // Puerto' );
  wrln( uri+' // URI' );
  wrln( usuario_http+' // Usuario_http' );
  wrln( clave_http+' // Clave_http' );
  wrln( IntToStr( inodo ) +' // inodo' );
  wrln( carpeta_tmp+' // carpeta_tmp' );
  wrln( IntToStr( maxnucleos )+' // maxNucleos');
  wrln( IntToStr( maxtareas )+' // maxTareas');
  wrln( IntToStr( maxhoras )+' // maxHoras');
  f.Free;
end;

{ TCola }
constructor TCola.Create(nombre: string; nnucleos, maxhoras, maxtareas: integer);
begin
  self.nombre := nombre;
  self.nnucleos := nnucleos;
  self.maxhoras := maxhoras;
  self.maxtareas := maxtareas;
end;


constructor TRBT_OddFace_Simple.Create;
begin

//  usuario := GetEnvironmentVariable('USER');
//  home := GetEnvironmentVariable('HOME');
  yo := ParamStr(0);

  cfg:= TConfigRec.CreateFromArchi( getDir_Bin + 'oddface_simple.cfg' );
  usuario:= cfg.usuario_http;
  home_:= getDir_bin;
  aplicacion := '';

  colas := TList.Create;
  colas.add(TCola.Create('normal', cfg.maxnucleos, cfg.maxhoras, cfg.maxtareas));

  ncolas := colas.Count;

  if (ParamCount >= 1) then
    orden := ParamStr(1)
  else
    orden := '?';

  sql := 'SELECT now()';
  db := TDBrosxCon.Create( cfg.IpFija, cfg.Host, cfg.Puerto, cfg.uri);


  ds := DB.sql_query(sql);
  if (ds = nil) or (ds.nrows = 0) then
  begin
    ds.Free;
    DB.Free;
    writeln('Error de conectividad con la DB remota ');
    exit;
  end;
  //  writeln('La hora del DB remoto es: ', ds.fields[0].AsString);
end;

function TRBT_OddFace_Simple.exec(apl: string; params: array of string;
  var mls: TStringList; flgWaitOnExit: boolean = true ): string;
var
  s: ansistring;
  res: boolean;
begin

  if flgWaitOnExit then
    res:= RunCommandIndir('', apl, params, s, [poWaitOnExit])
  else
    res:= RunCommandIndir('', apl, params, s, []);

  if res then
  begin
    mls := TStringList.Create;
    mls.LineBreak := #10;
    mls.Text := s;
    if mls.Count > 0 then
      Result := mls[mls.Count - 1]
    else
      Result := '';
  end
  else
    raise Exception.Create('Error TRBT_OddFace_Clustger.exec, comando: ' + apl);
end;


function TRBT_OddFace_Simple.exec(comando: string; params: array of string; flgWaitOnExit: boolean = true ): string;
var
  mls: TStringList;
begin
  Result := exec(comando, params, mls, flgWaitOnExit );
  mls.Free;
end;

function TRBT_OddFace_Simple.daofs(a: array of string): TDAofString;
var
  res: TDAOfString;
  k: integer;
begin
  setlength(res, length(a));
  for k := 0 to high(res) do
    res[k] := a[k];
  Result := res;
end;


procedure TRBT_OddFace_Simple.wrln(s: string);
begin
  if (verboso > 0) then
    writeln(s);
end;


procedure TRBT_OddFace_Simple.wrln_err(s: string);
begin
  writeln(s);
end;


(*
  nexval( $key, $reserva = 1) , retorna un identificador �nico global
  el par�metro $reserva indica la cantidad de identificadores consecutivos
  reservados (es decir asigandos y que no volveran ser asignados.
 *)
function TRBT_OddFace_Simple.nextval(key: string; reserva: integer = 1): integer;
var
  f: textfile;
  res, resx: integer;
  archi: string;
begin
  archi := home_ + 'nextval_' + key;
  if (fileexists(archi)) then
  begin
    assignfile(f, archi);
    reset(f);
    readln(f, res);
    closefile(f);
  end
  else
    res := 1;

  resx := res + reserva;

  assignfile(f, archi);
  rewrite(f);
  writeln(f, resx);
  closefile(f);
  Result := res;
end;






procedure TRBT_OddFace_Simple.remover_crontab;
var
  m: TStringList;
  ul: string;
begin
  ul := exec( crontab_cmd__, ['/delete', '/tn', '"rbt_oddface_simple"', '/F'], m);
  m.Free;
end;



procedure TRBT_OddFace_Simple.instalar_crontab;

var
  ul: string;
  m: TStringList;
begin
  ul := exec( crontab_cmd__, [
   '/create', '/sc', 'minute', '/mo', '3', '/tn', '"rbt_oddface_simple"',
   '/tr', getDir_Bin+'rbt_oddface_simple.exe'], m);
  m.Free;
end;


function TRBT_OddFace_Simple.substr(const s: string; k_base_0: integer;
  cnt: integer): string;
begin
  if cnt > 0 then
    Result := copy(s, k_base_0 + 1, cnt)
  else
    Result := copy(s, k_base_0 + 1, length(s) - k_base_0);
end;

procedure TRBT_OddFace_Simple.Ayuda;
begin
  if orden <> '' then
    writeln('orden desconocida: ' + orden);
  writeln('La sintaxis es ./rbt_OddFace_Simple orden [par2]');
  writeln('  Las ordenes conocidas son: ');
  writeln('    "run" que testea el estado del nodo y lanza o para aplicaciones');
  writeln('    "parar_tareas n" que manda parar "n" tareas. ');
  writeln('    "parar_crontab" elimina el crontab del usuario. ');
  writeln('    "instalar_crontab" instala el crontab del usuario. ');
  writeln('    "parar_todo" para las tareas y borra el crontab ');
  writeln('    "lst" resume cantidad de trabajos en ejecución. ');
  writeln('    "cancelar_trabajos" cancela todos los trabjos de usuario. ');
end;

procedure TRBT_OddFace_Simple.orden_lst;
var
  nEsperandoTodos, nEsperandoPropios, nTotalTodos, nTotalPropios: integer;
  kCola: integer;
  xcola: TCola;
  m: TStringList;
  k: integer;
begin
  for kCola := 0 to colas.Count - 1 do
  begin
    xcola := colas[kCola];
    writeln('Cola: ', xcola.nombre, ' ============================');
     s := exec('tasklist', ['/fi', '"imagename eq oddface_pig*"'], m);
     for k:= 0 to m.Count -1 do
      writeln( k, ': ', m[k ] );

    writeln('Total: ', m.Count-3);
    m.Free;
  end;
end;




procedure TRBT_OddFace_Simple.orden_cancelartrabajos(
  username, xcola, wk_nidProblema: string);
var
  s: string;
  m: TStringList;
  k: integer;
  job_id: string;
  name: string;
begin
  s := exec('tasklist', ['/fi', '"imagename eq oddface_pig*"'], m);
  (*
  for k:= 0 to m.Count -1 do
   writeln( k, ': ', m[k ] );
    *)
  for k := 3 to m.Count-1 do
  begin
    s := m[k];
    name:= getNextPalSep( s, ' ');
    job_id := getNextPalSep( s, ' ' );
    exec( 'taskkill', [ '/F', '/PID', job_id]  );
  end;
  Writeln('se cancelaron ', m.Count-3, ' trabajos' );
end;


procedure TRBT_OddFace_Simple.parar(ntareas: integer);
var
  sql: string;
begin
  sql := 'UPDATE ofe_cnt_terminar SET cnt = ' + IntToStr(ntareas) + ' WHERE 1';
  DB.Exec(sql);
end;



procedure TRBT_OddFace_Simple.runLanzarParar(xcola: TCola);
var
  nRunning: integer;
  margen: integer;
begin
  nRunning := getRunning;
  margen  := xcola.maxtareas - nRunning;
  if margen > 0 then
      lanzar( min(margen, ntares_por_vez), xcola);
end;



procedure TRBT_OddFace_Simple.lanzar(ntareas: integer; xcola: TCola);
var
  kbase, k, selk, nrows: integer;
  rec: TDataRecord;
  id_problema, apl_problema: string;
  aplicacion: string;
  idEjecutor: integer;
  idTrabajo: string;
  archi: string;
  fp: textfile;
  r, ul: string;
begin
  kbase := nextval('ejecutores', ntareas);
  nrows := ds.nrows;
  for  k := 0 to ntareas - 1 do
  begin
    selk := random(nrows);
    ds.go(selk);
    rec := ds.getFilaActiva;
    id_problema := trim(rec.fields[0].AsString);
    apl_problema := trim(rec.fields[1].AsString);
    {$IFDEF WINDOWS}
    if pos( '.exe', apl_problema ) = 0 then
      apl_problema:= apl_problema+'.exe';
    {$ENDIF}
    aplicacion := getDir_Bin + apl_problema;

    wrln('nrows: ' + IntToStr(nrows) + ' , selk: ' + IntToStr(selk) +
      ' , apl_problema: ' + apl_problema + ' , id_problema: ' + id_problema);

    if ((apl_problema <> '') and fileexists(aplicacion)) then
    begin
      idEjecutor := kbase + k;

        r := aplicacion + ' ' + IntToStr(idEjecutor) + ' ' + id_problema;
        LanzarApl( aplicacion, [ IntToStr(idEjecutor), id_problema ] );

      writeln('idEjecutor: ' + IntToStr(idEjecutor)  );
      sleep(pausa_segs * 1000);
    end;
  end;
end;


(*
  getRunning() retorna la cantidad de tareas ejecutando
 *)
function TRBT_OddFace_Simple.getRunning: integer;
var
  s: string;
  m: TStringList;
  n: integer;
begin
  s := exec('tasklist', ['/fi', '"imagename eq oddface_pig*"'], m);
  n := m.Count;
  m.Free;
  result := max( 0, n-3 );
end;




procedure TRBT_OddFace_Simple.orden_run;
var
  kcola: integer;
  sql: string;
begin
  sql := 'SELECT ofe_problemas.nid, ofe_tipos_problema.apl ' +
    'FROM ofe_problemas LEFT JOIN ofe_tipos_problema ON ofe_problemas.tipo = ofe_tipos_problema.nid '
    + ' JOIN ofe_grupo_ucluster ON ofe_problemas.grupo = ofe_grupo_ucluster.grupo ' +
    ' WHERE ofe_grupo_ucluster.ucluster = ''' + Usuario + ''' AND ofe_problemas.estado > 0 ';
  ds := DB.sql_query(sql);
  if (ds.nrows = 0) then
  begin
    wrln('No problems to solve.');
    exit;
  end;

  for kcola := 0 to ncolas - 1 do
  begin
    runLanzarParar(colas[kcola]);
  end;
  exit;
end;


procedure TRBT_OddFace_Simple.order_parar_tareas;
var
  nTareas: integer;
begin
  if (ParamCount > 1) then
    nTareas := StrToInt(ParamStr(2))
  else
    nTareas := 10000; //para todas

  parar(nTareas);
  exit;
end;

procedure TRBT_OddFace_Simple.dispatch_cmd;
var
  NombreCola, nHilos, email: string;
  k: integer;
  id_problema, apl_problema: string;
  aplicacion: string;
  idEjecutor: integer;
  idTrabajo: string;
  archi: string;
  fp: textfile;
  r, ul: string;
  xCola: TCola;
  buscando: boolean;
begin
  id_problema := '0';
  email := ParamStr(1);
  nombreCola := ParamStr(2);
  nHilos := ParamStr(3);
  if StrToInt(nHilos) > 40 then
  begin
    writeln('El máximo de nHilos es 40 en el CLusterUY');
    exit;
  end;
  apl_Problema := ParamStr(4);

  buscando := True;
  for k := 0 to colas.Count - 1 do
  begin
    xCola := colas[k];
    if xCola.nombre = nombreCola then
    begin
      buscando := False;
      break;
    end;
  end;

  if buscando then
  begin
    writeln('Nombre de cola inválido: ', nombreCola);
    exit;
  end;

  idEjecutor := nextval('ejecutores', 1);

  aplicacion := getDir_Bin + apl_Problema;
  if not fileexists(aplicacion) then
  begin
    writeln('No existe la aplciación: ' + aplicacion);
    exit;
  end;

  LanzarApl( aplicacion , [IntToStr(idEjecutor), id_problema ] )
end;


function TRBT_OddFace_Simple.ProcesarOrden: boolean;
var
  aCola: TCola;
  kCola: integer;
  wk_NidProblema: string;

begin
  if (orden = 'cancelar_trabajos') then
  begin
    if ParamCount > 1 then
      wk_NidProblema := 'wk_' + ParamStr(2)
    else
      wk_NidProblema := '';

    for kCola := 0 to Colas.Count - 1 do
    begin
      aCola := Colas[kCola];
      orden_cancelartrabajos(usuario, aCola.nombre, wk_NidProblema);
    end;

    Result := True;
    exit;
  end;


  if (orden = 'lst') then
  begin
    orden_lst;
    Result := True;
    exit;
  end;


  if (orden = 'run') then
  begin
    orden_run;
    Result := True;
    exit;
  end;

  if (orden = 'parar_tareas') then
  begin
    order_parar_tareas;
    Result := True;
    exit;
  end;

  if (orden = 'parar_crontab') then
  begin
    remover_crontab();
    Result := True;
    exit;
  end;

  if (orden = 'parar_todo') then
  begin
    parar(199000);
    remover_crontab();
    Result := True;
    exit;
  end;

  if (orden = 'instalar_crontab') then
  begin
    parar(0);
    instalar_crontab();
    Result := True;
    exit;
  end;

  Result := False;
end;


end.
