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