/* bench_connector.c — proc connector (netlink) y taskstats contra sondear /proc.
 *
 * Cierra el experimento 1 de SDD 25 §9, que quedo sin medir por falta de CAP_NET_ADMIN.
 * Requiere root. Tres fases:
 *   A. LATENCIA   fork+_exit+waitpid  vs  fork+_exit+esperar el evento del connector.
 *   B. CEGUERA    procesos cortos vistos por el connector vs vistos sondeando /proc cada 100 ms.
 *   C. TASKSTATS  una consulta binaria por genetlink vs parsear /proc/<pid>/stat.
 *
 * Se compila con: cc -O2 -o bench_connector bench_connector.c
 */
#define _GNU_SOURCE
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <fcntl.h>
#include <time.h>
#include <errno.h>
#include <dirent.h>
#include <sys/socket.h>
#include <sys/wait.h>
#include <sys/resource.h>
#include <poll.h>
#include <linux/netlink.h>
#include <linux/connector.h>
#include <linux/cn_proc.h>
#include <linux/genetlink.h>
#include <linux/taskstats.h>

static volatile long sink;
static inline double now_ns(void){ struct timespec t; clock_gettime(CLOCK_MONOTONIC,&t); return t.tv_sec*1e9+t.tv_nsec; }
static int cmpd(const void*a,const void*b){ double x=*(const double*)a,y=*(const double*)b; return (x>y)-(x<y); }
static void report(const char*n,double*r,int c,long it){ qsort(r,c,sizeof(double),cmpd);
  printf("%-44s %10.1f %10.1f   (n=%ld)\n",n,r[0],r[c/2],it); fflush(stdout); }
#define ROUNDS 7

/* ---------------------------------------------------------------- proc connector */

static int cn_open(void){
  int s = socket(PF_NETLINK, SOCK_DGRAM|SOCK_CLOEXEC, NETLINK_CONNECTOR);
  if(s<0) return -1;
  struct sockaddr_nl a; memset(&a,0,sizeof a);
  a.nl_family=AF_NETLINK; a.nl_groups=CN_IDX_PROC; a.nl_pid=getpid();
  if(bind(s,(struct sockaddr*)&a,sizeof a)<0){ close(s); return -1; }

  struct {
    struct nlmsghdr nl;
    struct cn_msg cn;
    enum proc_cn_mcast_op op;
  } __attribute__((packed)) m;
  memset(&m,0,sizeof m);
  m.nl.nlmsg_len=sizeof m; m.nl.nlmsg_pid=getpid(); m.nl.nlmsg_type=NLMSG_DONE;
  m.cn.id.idx=CN_IDX_PROC; m.cn.id.val=CN_VAL_PROC; m.cn.len=sizeof(enum proc_cn_mcast_op);
  m.op=PROC_CN_MCAST_LISTEN;
  if(send(s,&m,sizeof m,0)<0){ close(s); return -1; }
  /* El multicast del connector NO reintenta: si el buffer se llena, el evento se PIERDE (ENOBUFS)
   * y el que espera ese pid se queda esperando para siempre. Se agranda el buffer y, aun asi, la
   * espera lleva plazo (abajo). Es una propiedad del mecanismo, no un detalle de este programa. */
  int rb=4*1024*1024; setsockopt(s,SOL_SOCKET,SO_RCVBUFFORCE,&rb,sizeof rb);
  return s;
}

/* Lee un evento. Devuelve el proc_event, o NULL si el mensaje no lo era. */
static struct proc_event *cn_read(int s, char *buf, size_t n){
  ssize_t r = recv(s, buf, n, 0);
  if(r<=0) return NULL;
  struct nlmsghdr *nlh=(struct nlmsghdr*)buf;
  if(!NLMSG_OK(nlh,(size_t)r)) return NULL;
  struct cn_msg *cn=(struct cn_msg*)NLMSG_DATA(nlh);
  return (struct proc_event*)cn->data;
}

/* Vacia lo que haya en el socket sin bloquear. */
static void cn_drain(int s){
  int fl=fcntl(s,F_GETFL,0); fcntl(s,F_SETFL,fl|O_NONBLOCK);
  char b[4096]; while(recv(s,b,sizeof b,0)>0){}
  fcntl(s,F_SETFL,fl);
}

static void fase_a(int s){
  printf("\n### A. Latencia de notificacion: waitpid contra evento del connector\n");
  printf("%-44s %10s %10s\n","BENCH (ns por proceso)","min","mediana");
  const long N=200;

  double r[ROUNDS];
  for(int q=0;q<ROUNDS;q++){
    double t0=now_ns();
    for(long i=0;i<N;i++){ pid_t p=fork(); if(p==0) _exit(0); int st; waitpid(p,&st,0); }
    r[q]=(now_ns()-t0)/(double)N;
  }
  report("fork+_exit+waitpid (SIGCHLD, la base)",r,ROUNDS,N);

  char buf[8192];
  long perdidos=0;
  for(int q=0;q<ROUNDS;q++){
    cn_drain(s);
    double t0=now_ns();
    for(long i=0;i<N;i++){
      pid_t p=fork(); if(p==0) _exit(0);
      double lim=now_ns()+2e8;              /* 200 ms de plazo: si no llega, se PERDIO */
      for(;;){
        if(now_ns()>lim){ perdidos++; break; }
        struct pollfd pf={.fd=s,.events=POLLIN};
        if(poll(&pf,1,50)<=0) continue;
        struct proc_event *e=cn_read(s,buf,sizeof buf);
        if(!e) continue;
        if(e->what==PROC_EVENT_EXIT && e->event_data.exit.process_pid==p) break;
      }
      int st; waitpid(p,&st,0);
    }
    r[q]=(now_ns()-t0)/(double)N;
  }
  report("fork+_exit+esperar EXIT del connector",r,ROUNDS,N);
  printf("eventos EXIT que no llegaron en 200 ms: %ld de %ld\n",perdidos,(long)N*ROUNDS);
}

/* ---------------------------------------------------------------- ceguera del sondeo */

#define BURST 200        /* procesos cortos */
#define SPACING_US 10000 /* uno cada 10 ms => 2 s de rafaga */
#define POLL_MS 100      /* periodo del sondeo, generoso: un supervisor real no baja de aca */

/* Lanza BURST hijos que viven lo minimo, y escribe sus pids por el pipe. */
static void spawner(int wfd){
  for(int i=0;i<BURST;i++){
    pid_t p=fork();
    if(p==0) _exit(0);
    (void)!write(wfd,&p,sizeof p);
    int st; waitpid(p,&st,0);
    usleep(SPACING_US);
  }
  close(wfd);
  _exit(0);
}

static int en_lista(pid_t *v,int n,pid_t p){ for(int i=0;i<n;i++) if(v[i]==p) return 1; return 0; }

static double cpu_ms(void){
  struct rusage u; getrusage(RUSAGE_SELF,&u);
  return (u.ru_utime.tv_sec+u.ru_stime.tv_sec)*1e3
       + (u.ru_utime.tv_usec+u.ru_stime.tv_usec)/1e3;
}

static void fase_b(int s){
  printf("\n### B. Ceguera: %d procesos de vida minima, uno cada %d ms\n",BURST,SPACING_US/1000);

  /* --- observador 1: sondeo de /proc cada POLL_MS --- */
  int fd[2]; if(pipe(fd)) { perror("pipe"); return; }
  pid_t sp=fork();
  if(sp==0){ close(fd[0]); spawner(fd[1]); }
  close(fd[1]);
  static pid_t lanzados[BURST]; int nl=0;
  static pid_t vistos[BURST*8]; int nv=0;
  double c0=cpu_ms(), t0=now_ns(); int barridos=0;
  while(now_ns()-t0 < (double)BURST*SPACING_US*1000.0 + 3e8){
    DIR *d=opendir("/proc"); struct dirent *e;
    while(d && (e=readdir(d))){
      if(e->d_name[0]<'0'||e->d_name[0]>'9') continue;
      pid_t p=(pid_t)atol(e->d_name);
      if(nv<BURST*8 && !en_lista(vistos,nv,p)) vistos[nv++]=p;
    }
    if(d) closedir(d);
    barridos++;
    usleep(POLL_MS*1000);
  }
  double cpu_sondeo=cpu_ms()-c0;
  { int st; waitpid(sp,&st,0); }
  { pid_t p; while(read(fd[0],&p,sizeof p)==sizeof p) if(nl<BURST) lanzados[nl++]=p; }
  close(fd[0]);
  int cazados=0; for(int i=0;i<nl;i++) if(en_lista(vistos,nv,lanzados[i])) cazados++;
  printf("sondeo /proc cada %d ms : %d de %d procesos vistos  (%d barridos, %.1f ms de CPU)\n",
         POLL_MS,cazados,nl,barridos,cpu_sondeo);

  /* --- observador 2: connector --- */
  if(pipe(fd)){ perror("pipe"); return; }
  cn_drain(s);
  sp=fork();
  if(sp==0){ close(fd[0]); spawner(fd[1]); }
  close(fd[1]);
  nl=0; nv=0;
  c0=cpu_ms(); t0=now_ns();
  int flags=fcntl(s,F_GETFL,0); fcntl(s,F_SETFL,flags|O_NONBLOCK);
  char buf[8192];
  while(now_ns()-t0 < (double)BURST*SPACING_US*1000.0 + 3e8){
    struct proc_event *e=cn_read(s,buf,sizeof buf);
    if(!e){ usleep(200); continue; }
    if(e->what==PROC_EVENT_EXIT){
      pid_t p=e->event_data.exit.process_pid;
      if(nv<BURST*8 && !en_lista(vistos,nv,p)) vistos[nv++]=p;
    }
  }
  fcntl(s,F_SETFL,flags);
  double cpu_conn=cpu_ms()-c0;
  { int st; waitpid(sp,&st,0); }
  { pid_t p; while(read(fd[0],&p,sizeof p)==sizeof p) if(nl<BURST) lanzados[nl++]=p; }
  close(fd[0]);
  cazados=0; for(int i=0;i<nl;i++) if(en_lista(vistos,nv,lanzados[i])) cazados++;
  printf("proc connector (eventos)  : %d de %d procesos vistos  (%.1f ms de CPU)\n",
         cazados,nl,cpu_conn);
  printf("NOTA: el connector se lee aca con espera activa de 200 us, que NO es el modo de un\n"
         "      supervisor real (seria epoll) => su CPU es cota superior. Lo que cuenta: COBERTURA.\n");
}

/* ---------------------------------------------------------------- taskstats */

static int genl_family(int s,const char *name){
  struct { struct nlmsghdr n; struct genlmsghdr g; char buf[256]; } req;
  memset(&req,0,sizeof req);
  req.n.nlmsg_type=GENL_ID_CTRL; req.n.nlmsg_flags=NLM_F_REQUEST; req.n.nlmsg_seq=1; req.n.nlmsg_pid=getpid();
  req.g.cmd=CTRL_CMD_GETFAMILY; req.g.version=1;
  struct nlattr *a=(struct nlattr*)req.buf;
  a->nla_type=CTRL_ATTR_FAMILY_NAME; a->nla_len=NLA_HDRLEN+strlen(name)+1;
  strcpy((char*)a+NLA_HDRLEN,name);
  req.n.nlmsg_len=NLMSG_LENGTH(GENL_HDRLEN)+NLA_ALIGN(a->nla_len);
  if(send(s,&req,req.n.nlmsg_len,0)<0) return -1;

  char rb[4096]; ssize_t r=recv(s,rb,sizeof rb,0);
  if(r<0) return -1;
  struct nlmsghdr *nh=(struct nlmsghdr*)rb;
  if(nh->nlmsg_type==NLMSG_ERROR) return -1;
  struct nlattr *na=(struct nlattr*)((char*)NLMSG_DATA(nh)+GENL_HDRLEN);
  int len=NLMSG_PAYLOAD(nh,0)-GENL_HDRLEN;
  while(len>0){
    if(na->nla_type==CTRL_ATTR_FAMILY_ID) return *(__u16*)((char*)na+NLA_HDRLEN);
    len-=NLA_ALIGN(na->nla_len);
    na=(struct nlattr*)((char*)na+NLA_ALIGN(na->nla_len));
  }
  return -1;
}

static int ts_query(int s,int fam,pid_t pid,unsigned long long *cpu_ns){
  struct { struct nlmsghdr n; struct genlmsghdr g; char buf[128]; } req;
  memset(&req,0,sizeof req);
  req.n.nlmsg_type=fam; req.n.nlmsg_flags=NLM_F_REQUEST; req.n.nlmsg_seq=2; req.n.nlmsg_pid=getpid();
  req.g.cmd=TASKSTATS_CMD_GET; req.g.version=TASKSTATS_VERSION;
  struct nlattr *a=(struct nlattr*)req.buf;
  a->nla_type=TASKSTATS_CMD_ATTR_PID; a->nla_len=NLA_HDRLEN+sizeof(__u32);
  *(__u32*)((char*)a+NLA_HDRLEN)=pid;
  req.n.nlmsg_len=NLMSG_LENGTH(GENL_HDRLEN)+NLA_ALIGN(a->nla_len);
  if(send(s,&req,req.n.nlmsg_len,0)<0) return -1;

  char rb[8192]; ssize_t r=recv(s,rb,sizeof rb,0);
  if(r<0) return -1;
  struct nlmsghdr *nh=(struct nlmsghdr*)rb;
  if(nh->nlmsg_type==NLMSG_ERROR) return -1;
  struct nlattr *na=(struct nlattr*)((char*)NLMSG_DATA(nh)+GENL_HDRLEN);
  int len=NLMSG_PAYLOAD(nh,0)-GENL_HDRLEN;
  while(len>0){
    if(na->nla_type==TASKSTATS_TYPE_AGGR_PID || na->nla_type==TASKSTATS_TYPE_AGGR_TGID){
      struct nlattr *in=(struct nlattr*)((char*)na+NLA_HDRLEN);
      int il=na->nla_len-NLA_HDRLEN;
      while(il>0){
        if(in->nla_type==TASKSTATS_TYPE_STATS){
          struct taskstats *ts=(struct taskstats*)((char*)in+NLA_HDRLEN);
          *cpu_ns=ts->ac_utime*1000ULL+ts->ac_stime*1000ULL;
          return 0;
        }
        il-=NLA_ALIGN(in->nla_len);
        in=(struct nlattr*)((char*)in+NLA_ALIGN(in->nla_len));
      }
    }
    len-=NLA_ALIGN(na->nla_len);
    na=(struct nlattr*)((char*)na+NLA_ALIGN(na->nla_len));
  }
  return -1;
}

static void fase_c(void){
  printf("\n### C. Estadisticas de UNA tarea: taskstats binario contra /proc de texto\n");
  int s=socket(PF_NETLINK,SOCK_DGRAM|SOCK_CLOEXEC,NETLINK_GENERIC);
  if(s<0){ printf("taskstats: sin socket genetlink (%s)\n",strerror(errno)); return; }
  struct sockaddr_nl a; memset(&a,0,sizeof a); a.nl_family=AF_NETLINK; a.nl_pid=getpid();
  if(bind(s,(struct sockaddr*)&a,sizeof a)<0){ printf("taskstats: bind (%s)\n",strerror(errno)); close(s); return; }
  int fam=genl_family(s,TASKSTATS_GENL_NAME);
  if(fam<0){ printf("taskstats: la familia %s no esta (kernel sin TASKSTATS)\n",TASKSTATS_GENL_NAME); close(s); return; }
  unsigned long long ns=0;
  if(ts_query(s,fam,getpid(),&ns)<0){ printf("taskstats: la consulta fallo (%s)\n",strerror(errno)); close(s); return; }
  printf("familia TASKSTATS = %d, cpu de este proceso = %llu ns\n",fam,ns);

  printf("%-44s %10s %10s\n","BENCH (ns por consulta)","min","mediana");
  const long N=2000;
  double r[ROUNDS];
  for(int q=0;q<ROUNDS;q++){
    double t0=now_ns();
    for(long i=0;i<N;i++){ unsigned long long v; sink=ts_query(s,fam,getpid(),&v); }
    r[q]=(now_ns()-t0)/(double)N;
  }
  report("taskstats por genetlink (binario)",r,ROUNDS,N);

  for(int q=0;q<ROUNDS;q++){
    double t0=now_ns();
    for(long i=0;i<N;i++){
      int fd=open("/proc/self/stat",O_RDONLY); char b[2048];
      ssize_t n=read(fd,b,sizeof b-1); close(fd);
      if(n>0){ b[n]=0; char *p=strrchr(b,')'); for(int f=0;p&&f<12;f++) p=strchr(p+1,' ');
               sink=p?atol(p+1):0; }
    }
    r[q]=(now_ns()-t0)/(double)N;
  }
  report("open+read+parse /proc/self/stat",r,ROUNDS,N);
  close(s);
}

int main(void){
  if(geteuid()!=0) fprintf(stderr,"aviso: sin root el connector no se suscribe (CAP_NET_ADMIN)\n");
  printf("# bench_connector — SDD 25 seccion 9, experimento 1\n");
  int s=cn_open();
  if(s<0){ printf("proc connector: no disponible (%s)\n",strerror(errno)); }
  else { fase_a(s); fase_b(s); close(s); }
  fase_c();
  return 0;
}