server/increment.pc

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