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       name = xmalloc(0);
 120       id = atoi(before[5]);
 121       if (!strncmp(before[4], "USER", 4))
 122         {
 123           id_to_name(id, USERS_TABLE, &name);
 124           strcpy(before[5], name);
 125         }
 126       else if (!strncmp(before[4], "LIST", 4))
 127         {
 128           id_to_name(id, LIST_TABLE, &name);
 129           strcpy(before[5], name);
 130         }
 131       else if (!strncmp(before[4], "KERBEROS", 8))
 132         {
 133           id_to_name(id, STRINGS_TABLE, &name);
 134           strcpy(before[5], name);
 135         }
 136       break;
 137     case MCMAP_TABLE:
 138       strcpy(before[0], argv[0]);
 139       strcpy(before[1], argv[1]);
 140       beforec = 2;
 141       break;
 142     case MCNTMAP_TABLE:
 143       strcpy(before[0], argv[0]);
 144       strcpy(before[1], argv[1]);
 145       name_to_id(before[0], MACHINE_TABLE, &id);
 146       sprintf(before[2], "%d", id);
 147       name_to_id(before[1], CONTAINERS_TABLE, &id);
 148       sprintf(before[3], "%d", id);
 149       beforec = 4;
 150       break;
 151     case SVC_TABLE:
 152       strcpy(before[0], argv[0]);
 153       strcpy(before[1], argv[1]);
 154       strcpy(before[2], argv[2]);
 155       beforec = 3;
 156       break;
 157     case FILESYS_TABLE:
 158       sprintf(stmt_buf, "SELECT fs.label, fs.type, fs.mach_id, fs.name, "
 159               "fs.mount, fs.rwaccess, fs.comments, fs.owner, fs.owners, "
 160               "fs.createflg, fs.lockertype, fs.filsys_id FROM filesys fs "
 161               "WHERE %s", qual);
 162       dosql(before);
 163       name = xmalloc(0);
 164       id = atoi(before[2]);
 165       id_to_name(id, MACHINE_TABLE, &name);
 166       strcpy(before[2], name);
 167       id = atoi(before[7]);
 168       id_to_name(id, USERS_TABLE, &name);
 169       strcpy(before[7], name);
 170       id = atoi(before[8]);
 171       id_to_name(id, LIST_TABLE, &name);
 172       strcpy(before[8], name);
 173       free(name);
 174       beforec = 12;
 175       break;
 176     case QUOTA_TABLE:
 177       strcpy(before[0], "?");
 178       strcpy(before[1], argv[1]);
 179       strcpy(before[2], "?");
 180       sprintf(stmt_buf, "SELECT q.quota, fs.name FROM quota q, filesys fs "
 181               "WHERE %s AND fs.filsys_id = q.filsys_id", qual);
 182       dosql(&(before[3]));
 183       strcpy(before[2], argv[1]);
 184       beforec = 5;
 185       break;
 186     case LIST_TABLE:
 187       sprintf(stmt_buf, "SELECT l.name, l.active, l.publicflg, l.hidden, "
 188               "l.maillist, l.grouplist, l.gid, l.acl_type, l.acl_id, "
 189               "l.description, l.list_id FROM list l WHERE %s", qual);
 190       dosql(before);
 191       beforec = 11;
 192       break;
 193     case IMEMBERS_TABLE:
 194       id = (int) argv[0];
 195       sprintf(stmt_buf, "SELECT active, publicflg, hidden, maillist, "
 196               "grouplist, gid FROM list WHERE list_id = %d", id);
 197       dosql(&(before[3]));
 198       name = xmalloc(0);
 199       id_to_name(id, LIST_TABLE, &name);
 200       name2 = xmalloc(0);
 201       strcpy(before[0], name);
 202       strcpy(before[1], argv[1]);
 203       id = (int) argv[2];
 204       beforec = 10;
 205       if (!strcmp(before[1], "USER"))
 206         {
 207           id_to_name(id, USERS_TABLE, &name2);
 208           EXEC SQL SELECT status, users_id INTO :before[9], :before[11]
 209             FROM users WHERE users_id = :id;
 210           EXEC SQL SELECT list_id INTO :before[10] FROM list
 211             WHERE name = :name;
 212           beforec = 12;
 213       }
 214       else if (!strcmp(before[1], "LIST"))
 215         {
 216           id_to_name(id, LIST_TABLE, &name2);
 217           EXEC SQL SELECT list_id INTO :before[9] FROM list
 218             WHERE name = :name;
 219           sprintf(before[10], "%d", id);
 220           beforec = 11;
 221         }
 222       else if (!strcmp(before[1], "STRING") || !strcmp(before[1], "KERBEROS"))
 223         {
 224           id_to_name(id, STRINGS_TABLE, &name2);
 225           EXEC SQL SELECT list_id INTO :before[9] FROM list
 226             WHERE name = :name;
 227         }
 228       else if (!strcmp(before[1], "MACHINE"))
 229         {
 230           id_to_name(id, MACHINE_TABLE, &name2);
 231           EXEC SQL SELECT list_id INTO :before[9] FROM list
 232             WHERE name = :name;
 233           sprintf(before[10], "%d", id);
 234           beforec = 11;
 235         }
 236       strcpy(before[2], name2);
 237       free(name);
 238       free(name2);
 239       break;
 240     default:
 241         /*
 242         com_err(whoami, 0, "requested incremental on unexpected table `%s'",
 243                 table_name[table]);
 244         */
 245       break;
 246     }
 247 }
 248
 249
 250 void incremental_clear_before(void)
 251 {
 252   beforec = 0;
 253 }
 254
 255
 256 /* add an element to the incremental queue for the changed row */
 257
 258 void incremental_after(enum tables table, char *qual, char **argv)
 259 {
 260   char *name, *name2;
 261   EXEC SQL BEGIN DECLARE SECTION;
 262   int id;
 263   EXEC SQL END DECLARE SECTION;
 264   struct iupdate *iu;
 265
 266   switch (table)
 267     {
 268     case USERS_TABLE:
 269       sprintf(stmt_buf, "SELECT u.login, u.unix_uid, u.shell, "
 270               "u.winconsoleshell, u.last, u.first, u.middle, u.status, "
 271               "u.clearid, u.type, u.users_id FROM users u WHERE %s", qual);
 272       dosql(after);
 273       afterc = 11;
 274       break;
 275     case MACHINE_TABLE:
 276       sprintf(stmt_buf, "SELECT m.name, m.vendor, m.mach_id FROM machine m "
 277               "WHERE %s", qual);
 278       dosql(after);
 279       afterc = 3;
 280       break;
 281     case CLUSTERS_TABLE:
 282       sprintf(stmt_buf, "SELECT c.name, c.description, c.location, "
 283               "c.clu_id FROM clusters c WHERE %s", qual);
 284       dosql(after);
 285       afterc = 4;
 286       break;
 287     case CONTAINERS_TABLE:
 288       sprintf(stmt_buf, "SELECT c.name, c.description, c.location, c.contact, "
 289               "c.acl_type, c.acl_id, c.cnt_id FROM containers c WHERE %s",
 290               qual);
 291       dosql(after);
 292       afterc = 7;
 293       name = xmalloc(0);
 294       id = atoi(after[5]);
 295       if (!strncmp(after[4], "USER", 4))
 296         {
 297           id_to_name(id, USERS_TABLE, &name);
 298           strcpy(after[5], name);
 299         }
 300       else if (!strncmp(after[4], "LIST", 4))
 301         {
 302           id_to_name(id, LIST_TABLE, &name);
 303           strcpy(after[5], name);
 304         }
 305       else if (!strncmp(after[4], "KERBEROS", 8))
 306         {
 307           id_to_name(id, STRINGS_TABLE, &name);
 308           strcpy(after[5], name);
 309         }
 310       break;
 311     case MCMAP_TABLE:
 312       strcpy(after[0], argv[0]);
 313       strcpy(after[1], argv[1]);
 314       afterc = 2;
 315       break;
 316     case MCNTMAP_TABLE:
 317       strcpy(after[0], argv[0]);
 318       strcpy(after[1], argv[1]);
 319       name_to_id(after[0], MACHINE_TABLE, &id);
 320       sprintf(after[2], "%d", id);
 321       name_to_id(after[1], CONTAINERS_TABLE, &id);
 322       sprintf(after[3], "%d", id);
 323       afterc = 4;
 324       break;
 325     case SVC_TABLE:
 326       strcpy(after[0], argv[0]);
 327       strcpy(after[1], argv[1]);
 328       strcpy(after[2], argv[2]);
 329       afterc = 3;
 330       break;
 331     case FILESYS_TABLE:
 332       sprintf(stmt_buf, "SELECT fs.label, fs.type, fs.mach_id, fs.name, "
 333               "fs.mount, fs.rwaccess, fs.comments, fs.owner, fs.owners, "
 334               "fs.createflg, fs.lockertype, fs.filsys_id FROM filesys fs "
 335               "WHERE %s", qual);
 336       dosql(after);
 337       name = xmalloc(0);
 338       id = atoi(after[2]);
 339       id_to_name(id, MACHINE_TABLE, &name);
 340       strcpy(after[2], name);
 341       id = atoi(after[7]);
 342       id_to_name(id, USERS_TABLE, &name);
 343       strcpy(after[7], name);
 344       id = atoi(after[8]);
 345       id_to_name(id, LIST_TABLE, &name);
 346       strcpy(after[8], name);
 347       free(name);
 348       afterc = 12;
 349       break;
 350     case QUOTA_TABLE:
 351       strcpy(after[0], "?");
 352       strcpy(after[1], argv[1]);
 353       strcpy(after[2], "?");
 354       sprintf(stmt_buf, "SELECT q.quota, fs.name FROM quota q, filesys fs "
 355               "WHERE %s and fs.filsys_id = q.filsys_id and q.type = '%s'",
 356               qual, argv[1]);
 357       dosql(&(after[3]));
 358       afterc = 5;
 359       break;
 360     case LIST_TABLE:
 361       sprintf(stmt_buf, "SELECT l.name, l.active, l.publicflg, l.hidden, "
 362               "l.maillist, l.grouplist, l.gid, l.acl_type, l.acl_id, "
 363               "l.description, l.list_id FROM list l WHERE %s", qual);
 364       dosql(after);
 365       afterc = 11;
 366       break;
 367     case IMEMBERS_TABLE:
 368       id = (int) argv[0];
 369       sprintf(stmt_buf, "SELECT active, publicflg, hidden, maillist, "
 370               "grouplist, gid FROM list WHERE list_id = %d", id);
 371       dosql(&(after[3]));
 372       name = xmalloc(0);
 373       id_to_name(id, LIST_TABLE, &name);
 374       name2 = xmalloc(0);
 375       strcpy(after[0], name);
 376       strcpy(after[1], argv[1]);
 377       id = (int) argv[2];
 378       afterc = 10;
 379       if (!strcmp(after[1], "USER"))
 380         {
 381           id_to_name(id, USERS_TABLE, &name2);
 382           EXEC SQL SELECT status, users_id INTO :after[9], :after[11]
 383             FROM users WHERE users_id = :id;
 384           EXEC SQL SELECT list_id INTO :after[10] FROM list
 385             WHERE name = :name;
 386           afterc = 12;
 387         }
 388       else if (!strcmp(after[1], "LIST"))
 389         {
 390           id_to_name(id, LIST_TABLE, &name2);
 391           EXEC SQL SELECT list_id INTO :after[9] FROM list
 392             WHERE name = :name;
 393           sprintf(after[10], "%d", id);
 394           afterc = 11;
 395         }
 396       else if (!strcmp(after[1], "STRING") || !strcmp(after[1], "KERBEROS"))
 397         {
 398           id_to_name(id, STRINGS_TABLE, &name2);
 399           EXEC SQL SELECT list_id INTO :after[9] FROM list
 400             WHERE name = :name;
 401         }
 402       else if (!strcmp(after[1], "MACHINE"))
 403         {
 404           id_to_name(id, MACHINE_TABLE, &name2);
 405           EXEC SQL SELECT list_id INTO :after[9] FROM list
 406             WHERE name = :name;
 407           sprintf(after[10], "%d", id);
 408           afterc = 11;
 409         }
 410       strcpy(after[2], name2);
 411       free(name);
 412       free(name2);
 413       break;
 414     case NO_TABLE:
 415       afterc = 0;
 416       table = beforetable;
 417       break;
 418     default:
 419         /*
 420         com_err(whoami, 0, "requested incremental on unexpected table `%s'",
 421                 table_name[table]);
 422         */
 423       break;
 424     }
 425
 426   iu = xmalloc(sizeof(struct iupdate));
 427   iu->table = table_name[table];
 428   iu->beforec = beforec;
 429   iu->before = copy_argv(before, beforec);
 430   iu->afterc = afterc;
 431   iu->after = copy_argv(after, afterc);
 432   sq_save_data(incremental_sq, iu);
 433 }
 434
 435 void incremental_clear_after(void)
 436 {
 437   incremental_after(NO_TABLE, NULL, NULL);
 438 }
 439
 440
 441 /* Called when the current transaction is committed to start any queued
 442  * incremental updates.  This caches the update table the first time it
 443  * is called.
 444  */
 445
 446 struct inc_cache {
 447   struct inc_cache *next;
 448   char *table, *service;
 449 };
 450
 451
 452 void incremental_update(void)
 453 {
 454   static int inited = 0;
 455   static struct inc_cache *cache;
 456   struct inc_cache *c;
 457   EXEC SQL BEGIN DECLARE SECTION;
 458   char tab[INCREMENTAL_TABLE_NAME_SIZE], serv[INCREMENTAL_SERVICE_SIZE];
 459   EXEC SQL END DECLARE SECTION;
 460   struct iupdate *iu, *iu_save;
 461
 462   if (!inited)
 463     {
 464       inited++;
 465
 466       EXEC SQL DECLARE inc CURSOR FOR SELECT table_name, service
 467         FROM incremental;
 468       EXEC SQL OPEN inc;
 469       while (1)
 470         {
 471           EXEC SQL FETCH inc INTO :tab, :serv;
 472           if (sqlca.sqlcode)
 473             break;
 474           c = xmalloc(sizeof(struct inc_cache));
 475           c->next = cache;
 476           c->table = xstrdup(strtrim(tab));
 477           c->service = xstrdup(strtrim(serv));
 478           cache = c;
 479         }
 480       EXEC SQL CLOSE inc;
 481       EXEC SQL COMMIT WORK;
 482     }
 483
 484   while (sq_remove_data(incremental_sq, &iu))
 485     {
 486       for (c = cache; c; c = c->next)
 487         {
 488           if (!strcmp(c->table, iu->table))
 489             {
 490               iu->service = c->service;
 491               iu_save = xmalloc(sizeof(struct iupdate));
 492               iu_save->service = iu->service;
 493               iu_save->table = iu->table;
 494               iu_save->beforec = iu->beforec;
 495               iu_save->afterc = iu->afterc;
 496               iu_save->before = copy_argv(iu->before, iu->beforec);
 497               iu_save->after = copy_argv(iu->after, iu->afterc);
 498               sq_save_data(incremental_exec, iu_save);
 499             }
 500         }
 501       if (!c)
 502         {
 503           free_argv(iu->before, iu->beforec);
 504           free_argv(iu->after, iu->afterc);
 505           free(iu);
 506         }
 507     }
 508   if (inc_running == 0)
 509     next_incremental();
 510 }
 511
 512 /* Pro*C 2.2.4 can't cope with the sigset_t below, at least in Solaris 2.6.
 513    We add DEFINE=_PROC_ to the proc invocation and then #ifndef that around
 514    this function so proc will pass it through without reading it. */
 515
 516 #ifndef _PROC_
 517 void next_incremental(void)
 518 {
 519   struct iupdate *iu;
 520   char *argv[MAXARGC * 2 + 4], cafter[3], cbefore[3], prog[MAXPATHLEN];
 521   int i;
 522   sigset_t sigs;
 523
 524   if (!incremental_exec)
 525     incremental_init();
 526
 527   if (sq_empty(incremental_exec) ||
 528       (inc_running && now - inc_started < INC_TIMEOUT))
 529     return;
 530
 531   if (inc_running)
 532     com_err(whoami, 0, "incremental timeout on pid %d", inc_pid);
 533
 534   sq_remove_data(incremental_exec, &iu);
 535   argv[1] = iu->table;
 536   sprintf(cbefore, "%d", iu->beforec);
 537   argv[2] = cbefore;
 538   sprintf(cafter, "%d", iu->afterc);
 539   argv[3] = cafter;
 540   for (i = 0; i < iu->beforec; i++)
 541     argv[4 + i] = iu->before[i];
 542   for (i = 0; i < iu->afterc; i++)
 543     argv[4 + iu->beforec + i] = iu->after[i];
 544
 545   sprintf(prog, "%s/%s.incr", BIN_DIR, iu->service);
 546   argv[0] = prog;
 547   argv[4 + iu->beforec + iu->afterc] = 0;
 548
 549   sigemptyset(&sigs);
 550   sigaddset(&sigs, SIGCHLD);
 551   sigprocmask(SIG_BLOCK, &sigs, NULL);
 552   inc_pid = vfork();
 553   switch (inc_pid)
 554     {
 555     case 0:
 556       execv(prog, argv);
 557       _exit(1);
 558     case -1:
 559       com_err(whoami, 0, "Failed to start incremental update");
 560       break;
 561     default:
 562       inc_running = 1;
 563       inc_started = now;
 564     }
 565   sigprocmask(SIG_UNBLOCK, &sigs, NULL);
 566
 567   free_argv(iu->before, iu->beforec);
 568   free_argv(iu->after, iu->afterc);
 569   free(iu);
 570 }
 571 #endif
 572
 573 /* Called when the current transaction is aborted to throw away any queued
 574  * incremental updates
 575  */
 576
 577 void incremental_flush(void)
 578 {
 579   struct iupdate *iu;
 580
 581   while (sq_get_data(incremental_sq, &iu))
 582     {
 583       free_argv(iu->before, iu->beforec);
 584       free_argv(iu->after, iu->afterc);
 585       free(iu);
 586     }
 587   sq_destroy(incremental_sq);
 588   incremental_sq = sq_create();
 589 }
 590
 591
 592 char **copy_argv(char **argv, int argc)
 593 {
 594   char **ret = xmalloc(sizeof(char *) * argc);
 595   while (--argc >= 0)
 596     ret[argc] = xstrdup(strtrim(argv[argc]));
 597   return ret;
 598 }
 599
 600 void free_argv(char **argv, int argc)
 601 {
 602   while (--argc >= 0)
 603     free(argv[argc]);
 604   free(argv);
 605 }
 606
 607 int table_num(char *name)
 608 {
 609   int i;
 610
 611   for (i = num_tables - 1; i; i--)
 612     {
 613       if (!strcmp(table_name[i], name))
 614         break;
 615     }
 616
 617   return i; /* 0 = "none" if no match */
 618 }