server/increment.pc

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