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