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