server/increment.pc

   1 /*
   2  *      $Source$
   3  *      $Author$
   4  *      $Header$
   5  *
   6  *      Copyright (C) 1989 by the Massachusetts Institute of Technology
   7  *      For copying and distribution information, please see the file
   8  *      <mit-copyright.h>.
   9  *
  10  */
  11
  12 #ifndef lint
  13 static char *rcsid_increment_dc = "$Header$";
  14 #endif lint
  15
  16 #include <mit-copyright.h>
  17 #include <moira.h>
  18 #include <fcntl.h>
  19 #include <unistd.h>
  20 #include <signal.h>
  21 #include "mr_server.h"
  22 #include "query.h"
  23 #include "qrtn.h"
  24 EXEC SQL INCLUDE sqlca;
  25
  26 extern char *whoami;
  27 extern char *table_name[];
  28 extern int num_tables;
  29
  30 int inc_pid = 0;
  31 int inc_running = 0;
  32 time_t inc_started;
  33
  34 #define MAXARGC 15
  35
  36 EXEC SQL WHENEVER SQLERROR DO dbmserr();
  37
  38 /* structures to save before args */
  39 static char *before[MAXARGC];
  40 static int beforec;
  41 static enum tables beforetable;
  42
  43 /* structures to save after args */
  44 static char *after[MAXARGC];
  45 static int afterc;
  46
  47 /* structures to save entire sets of incremental changes */
  48 struct save_queue *incremental_sq = NULL;
  49 struct save_queue *incremental_exec = NULL;
  50 struct iupdate {
  51   char *table;
  52   int beforec;
  53   char **before;
  54   int afterc;
  55   char **after;
  56   char *service;
  57 };
  58
  59 void next_incremental(void);
  60 char **copy_argv(char **argv, int argc);
  61 void free_argv(char **argv, int argc);
  62 int table_num(char *table);
  63
  64 void incremental_init(void)
  65 {
  66   int i;
  67
  68   if (!incremental_sq)
  69     incremental_sq = sq_create();
  70   if (!incremental_exec)
  71     incremental_exec = sq_create();
  72
  73   for (i = 0; i < MAXARGC; i++)
  74     {
  75       before[i] = malloc(ARGLEN);
  76       after[i] = malloc(ARGLEN);
  77     }
  78 }
  79
  80
  81 /* record the state of a table row before it is changed */
  82
  83 void incremental_before(enum tables table, char *qual, char **argv)
  84 {
  85   EXEC SQL BEGIN DECLARE SECTION;
  86   int id;
  87   EXEC SQL END DECLARE SECTION;
  88
  89   char *name;
  90
  91   beforetable = table;
  92
  93   switch (table)
  94     {
  95     case USERS_TABLE:
  96       sprintf(stmt_buf, "SELECT u.login, u.unix_uid, u.shell, u.last, "
  97               "u.first, u.middle, u.status, u.clearid, u.type "
  98               "FROM users u WHERE %s", qual);
  99       dosql(before);
 100       beforec = 9;
 101       break;
 102     case MACHINE_TABLE:
 103       sprintf(stmt_buf, "SELECT m.name, m.vendor FROM machine m "
 104               "WHERE %s", qual);
 105       dosql(before);
 106       beforec = 2;
 107       break;
 108     case CLUSTER_TABLE:
 109       sprintf(stmt_buf, "SELECT c.name, c.description, c.location "
 110               "FROM clusters c WHERE %s", qual);
 111       dosql(before);
 112       beforec = 3;
 113       break;
 114     case MCMAP_TABLE:
 115       strcpy(before[0], argv[0]);
 116       strcpy(before[1], argv[1]);
 117       beforec = 2;
 118       break;
 119     case SVC_TABLE:
 120       strcpy(before[0], argv[0]);
 121       strcpy(before[1], argv[1]);
 122       strcpy(before[2], argv[2]);
 123       beforec = 3;
 124       break;
 125     case FILESYS_TABLE:
 126       sprintf(stmt_buf, "SELECT fs.label, fs.type, fs.mach_id, fs.name, "
 127               "fs.mount, fs.rwaccess, fs.comments, fs.owner, fs.owners, "
 128               "fs.createflg, fs.lockertype FROM filesys fs WHERE %s", qual);
 129       dosql(before);
 130       name = malloc(0);
 131       id = atoi(before[2]);
 132       id_to_name(id, MACHINE_TABLE, &name);
 133       strcpy(before[2], name);
 134       id = atoi(before[7]);
 135       id_to_name(id, USERS_TABLE, &name);
 136       strcpy(before[7], name);
 137       id = atoi(before[8]);
 138       id_to_name(id, LIST_TABLE, &name);
 139       strcpy(before[8], name);
 140       free(name);
 141       beforec = 11;
 142       break;
 143     case QUOTA_TABLE:
 144       strcpy(before[0], "?");
 145       strcpy(before[1], argv[1]);
 146       strcpy(before[2], "?");
 147       sprintf(stmt_buf, "SELECT q.quota, fs.name FROM quota q, filesys fs "
 148               "WHERE %s AND fs.filsys_id = q.filsys_id", qual);
 149       dosql(&(before[3]));
 150       strcpy(before[2], argv[1]);
 151       beforec = 5;
 152       break;
 153     case LIST_TABLE:
 154       sprintf(stmt_buf, "SELECT l.name, l.active, l.publicflg, l.hidden, "
 155               "l.maillist, l.grouplist, l.gid, l.acl_type, l.acl_id, "
 156               "l.description FROM list l WHERE %s", qual);
 157       dosql(before);
 158       beforec = 10;
 159       break;
 160     case IMEMBERS_TABLE:
 161       id = (int) argv[0];
 162       sprintf(stmt_buf, "SELECT active, publicflg, hidden, maillist, "
 163               "grouplist, gid FROM list WHERE list_id = %d", id);
 164       dosql(&(before[3]));
 165       name = malloc(0);
 166       id_to_name(id, LIST_TABLE, &name);
 167       strcpy(before[0], name);
 168       strcpy(before[1], argv[1]);
 169       id = (int) argv[2];
 170       beforec = 9;
 171       if (!strcmp(before[1], "USER"))
 172         {
 173           id_to_name(id, USERS_TABLE, &name);
 174           EXEC SQL SELECT status INTO :before[9] FROM users
 175             WHERE users_id = :id;
 176           beforec = 10;
 177       }
 178       else if (!strcmp(before[1], "LIST"))
 179         id_to_name(id, LIST_TABLE, &name);
 180       else if (!strcmp(before[1], "STRING") || !strcmp(before[1], "KERBEROS"))
 181         id_to_name(id, STRINGS_TABLE, &name);
 182       strcpy(before[2], name);
 183       free(name);
 184       break;
 185     default:
 186         /*
 187         com_err(whoami, 0, "requested incremental on unexpected table `%s'",
 188                 table_name[table]);
 189         */
 190       break;
 191     }
 192 }
 193
 194
 195 void incremental_clear_before(void)
 196 {
 197   beforec = 0;
 198 }
 199
 200
 201 /* add an element to the incremental queue for the changed row */
 202
 203 void incremental_after(enum tables table, char *qual, char **argv)
 204 {
 205   char *name;
 206   EXEC SQL BEGIN DECLARE SECTION;
 207   int id;
 208   EXEC SQL END DECLARE SECTION;
 209   struct iupdate *iu;
 210
 211   switch (table)
 212     {
 213     case USERS_TABLE:
 214       sprintf(stmt_buf, "SELECT u.login, u.unix_uid, u.shell, u.last, "
 215               "u.first, u.middle, u.status, u.clearid, u.type "
 216               "FROM users u WHERE %s", qual);
 217       dosql(after);
 218       afterc = 9;
 219       break;
 220     case MACHINE_TABLE:
 221       sprintf(stmt_buf, "SELECT m.name, m.vendor FROM machine m "
 222               "WHERE %s", qual);
 223       dosql(after);
 224       afterc = 2;
 225       break;
 226     case CLUSTER_TABLE:
 227       sprintf(stmt_buf, "SELECT c.name, c.description, c.location "
 228               "FROM clusters c WHERE %s", qual);
 229       dosql(after);
 230       afterc = 3;
 231       break;
 232     case MCMAP_TABLE:
 233       strcpy(after[0], argv[0]);
 234       strcpy(after[1], argv[1]);
 235       afterc = 2;
 236       break;
 237     case SVC_TABLE:
 238       strcpy(after[0], argv[0]);
 239       strcpy(after[1], argv[1]);
 240       strcpy(after[2], argv[2]);
 241       afterc = 3;
 242       break;
 243     case FILESYS_TABLE:
 244       sprintf(stmt_buf, "SELECT fs.label, fs.type, fs.mach_id, fs.name, "
 245               "fs.mount, fs.rwaccess, fs.comments, fs.owner, fs.owners, "
 246               "fs.createflg, fs.lockertype FROM filesys fs WHERE %s", qual);
 247       dosql(after);
 248       name = malloc(0);
 249       id = atoi(after[2]);
 250       id_to_name(id, MACHINE_TABLE, &name);
 251       strcpy(after[2], name);
 252       id = atoi(after[7]);
 253       id_to_name(id, USERS_TABLE, &name);
 254       strcpy(after[7], name);
 255       id = atoi(after[8]);
 256       id_to_name(id, LIST_TABLE, &name);
 257       strcpy(after[8], name);
 258       free(name);
 259       afterc = 11;
 260       break;
 261     case QUOTA_TABLE:
 262       strcpy(after[0], "?");
 263       strcpy(after[1], argv[1]);
 264       strcpy(after[2], "?");
 265       sprintf(stmt_buf, "SELECT q.quota, fs.name FROM quota q, filesys fs "
 266               "WHERE %s and fs.filsys_id = q.filsys_id and q.type = '%s'",
 267               qual, argv[1]);
 268       dosql(&(after[3]));
 269       afterc = 5;
 270       break;
 271     case LIST_TABLE:
 272       sprintf(stmt_buf, "SELECT l.name, l.active, l.publicflg, l.hidden, "
 273               "l.maillist, l.grouplist, l.gid, l.acl_type, l.acl_id, "
 274               "l.description FROM list l WHERE %s", qual);
 275       dosql(after);
 276       afterc = 10;
 277       break;
 278     case IMEMBERS_TABLE:
 279       id = (int) argv[0];
 280       sprintf(stmt_buf, "SELECT active, publicflg, hidden, maillist, "
 281               "grouplist, gid FROM list WHERE list_id = %d", id);
 282       dosql(&(after[3]));
 283       name = malloc(0);
 284       id_to_name(id, LIST_TABLE, &name);
 285       strcpy(after[0], name);
 286       strcpy(after[1], argv[1]);
 287       id = (int) argv[2];
 288       afterc = 9;
 289       if (!strcmp(after[1], "USER"))
 290         {
 291           id_to_name(id, USERS_TABLE, &name);
 292           EXEC SQL SELECT status INTO :after[9] FROM users
 293             WHERE users_id = :id;
 294           afterc = 10;
 295         }
 296       else if (!strcmp(after[1], "LIST"))
 297         id_to_name(id, LIST_TABLE, &name);
 298       else if (!strcmp(after[1], "STRING") || !strcmp(after[1], "KERBEROS"))
 299         id_to_name(id, STRINGS_TABLE, &name);
 300       strcpy(after[2], name);
 301       free(name);
 302       break;
 303     case NO_TABLE:
 304       afterc = 0;
 305       table = beforetable;
 306       break;
 307     default:
 308         /*
 309         com_err(whoami, 0, "requested incremental on unexpected table `%s'",
 310                 table_name[table]);
 311         */
 312       break;
 313     }
 314
 315   iu = malloc(sizeof(struct iupdate));
 316   iu->table = table_name[table];
 317   iu->beforec = beforec;
 318   iu->before = copy_argv(before, beforec);
 319   iu->afterc = afterc;
 320   iu->after = copy_argv(after, afterc);
 321   sq_save_data(incremental_sq, iu);
 322 }
 323
 324 void incremental_clear_after(void)
 325 {
 326   incremental_after(NO_TABLE, NULL, NULL);
 327 }
 328
 329
 330 /* Called when the current transaction is committed to start any queued
 331  * incremental updates.  This caches the update table the first time it
 332  * is called.
 333  */
 334
 335 struct inc_cache {
 336   struct inc_cache *next;
 337   char *table, *service;
 338 };
 339
 340
 341 void incremental_update(void)
 342 {
 343   static int inited = 0;
 344   static struct inc_cache *cache;
 345   struct inc_cache *c;
 346   EXEC SQL BEGIN DECLARE SECTION;
 347   char tab[17], serv[17];
 348   EXEC SQL END DECLARE SECTION;
 349   struct iupdate *iu;
 350
 351   if (!inited)
 352     {
 353       inited++;
 354
 355       EXEC SQL DECLARE inc CURSOR FOR SELECT table_name, service
 356         FROM incremental;
 357       EXEC SQL OPEN inc;
 358       while (1)
 359         {
 360           EXEC SQL FETCH inc INTO :tab, :serv;
 361           if (sqlca.sqlcode)
 362             break;
 363           c = malloc(sizeof(struct inc_cache));
 364           c->next = cache;
 365           c->table = strsave(strtrim(tab));
 366           c->service = strsave(strtrim(serv));
 367           cache = c;
 368         }
 369       EXEC SQL CLOSE inc;
 370       EXEC SQL COMMIT WORK;
 371     }
 372
 373   while (sq_remove_data(incremental_sq, &iu))
 374     {
 375       for (c = cache; c; c = c->next)
 376         {
 377           if (!strcmp(c->table, iu->table))
 378             {
 379               iu->service = c->service;
 380               sq_save_data(incremental_exec, iu);
 381             }
 382         }
 383     }
 384   if (inc_running == 0)
 385     next_incremental();
 386 }
 387
 388
 389 void next_incremental(void)
 390 {
 391   struct iupdate *iu;
 392   char *argv[MAXARGC * 2 + 4], cafter[3], cbefore[3], prog[BUFSIZ];
 393   int i;
 394   sigset_t sigs;
 395
 396   if (!incremental_exec)
 397     incremental_init();
 398
 399   if (sq_empty(incremental_exec) ||
 400       (inc_running && now - inc_started < INC_TIMEOUT))
 401     return;
 402
 403   if (inc_running)
 404     com_err(whoami, 0, "incremental timeout on pid %d", inc_pid);
 405
 406   sq_remove_data(incremental_exec, &iu);
 407   argv[1] = iu->table;
 408   sprintf(cbefore, "%d", iu->beforec);
 409   argv[2] = cbefore;
 410   sprintf(cafter, "%d", iu->afterc);
 411   argv[3] = cafter;
 412   for (i = 0; i < iu->beforec; i++)
 413     argv[4 + i] = iu->before[i];
 414   for (i = 0; i < iu->afterc; i++)
 415     argv[4 + iu->beforec + i] = iu->after[i];
 416
 417   sprintf(prog, "%s/%s.incr", BIN_DIR, iu->service);
 418   argv[0] = prog;
 419   argv[4 + iu->beforec + iu->afterc] = 0;
 420
 421   sigemptyset(&sigs);
 422   sigaddset(&sigs, SIGCHLD);
 423   sigprocmask(SIG_BLOCK, &sigs, NULL);
 424   inc_pid = vfork();
 425   switch (inc_pid)
 426     {
 427     case 0:
 428       execv(prog, argv);
 429       _exit(1);
 430     case -1:
 431       com_err(whoami, 0, "Failed to start incremental update");
 432       break;
 433     default:
 434       inc_running = 1;
 435       inc_started = now;
 436     }
 437   sigprocmask(SIG_UNBLOCK, &sigs, NULL);
 438
 439   free_argv(iu->before, iu->beforec);
 440   free_argv(iu->after, iu->afterc);
 441   free(iu);
 442 }
 443
 444
 445 /* Called when the current transaction is aborted to throw away any queued
 446  * incremental updates
 447  */
 448
 449 void incremental_flush(void)
 450 {
 451   struct iupdate *iu;
 452
 453   while (sq_get_data(incremental_sq, &iu))
 454     {
 455       free_argv(iu->before, iu->beforec);
 456       free_argv(iu->after, iu->afterc);
 457       free(iu);
 458     }
 459   sq_destroy(incremental_sq);
 460   incremental_sq = sq_create();
 461 }
 462
 463
 464 char **copy_argv(char **argv, int argc)
 465 {
 466   char **ret = malloc(sizeof(char *) * argc);
 467   while (--argc >= 0)
 468     ret[argc] = strsave(strtrim(argv[argc]));
 469   return ret;
 470 }
 471
 472 void free_argv(char **argv, int argc)
 473 {
 474   while (--argc >= 0)
 475     free(argv[argc]);
 476   free(argv);
 477 }
 478
 479 int table_num(char *name)
 480 {
 481   int i;
 482
 483   for (i = num_tables - 1; i; i--)
 484     {
 485       if (!strcmp(table_name[i], name))
 486         break;
 487     }
 488
 489   return i; /* 0 = "none" if no match */
 490 }