/* * $Source$ * $Author$ * $Header$ * * Copyright (C) 1989 by the Massachusetts Institute of Technology * For copying and distribution information, please see the file * . * */ #ifndef lint static char *rcsid_increment_dc = "$Header$"; #endif lint #include #include #include #include "query.h" #include "mr_server.h" EXEC SQL INCLUDE sqlca; extern char *whoami; char *malloc(); int inc_pid = 0; int inc_running = 0; time_t inc_started; #define MAXARGC 15 EXEC SQL WHENEVER SQLERROR CALL ingerr; /* structures to save before args */ static char beforeb[MAXARGC][ARGLEN]; static char *before[MAXARGC]; EXEC SQL BEGIN DECLARE SECTION; char *barg0, *barg1, *barg2, *barg3, *barg4; char *barg5, *barg6, *barg7, *barg8, *barg9; char *barg10, *barg11, *barg12, *barg13, *barg14; EXEC SQL END DECLARE SECTION; static int beforec; static char *beforetable; /* structures to save after args */ static char afterb[MAXARGC][ARGLEN]; static char *after[MAXARGC]; EXEC SQL BEGIN DECLARE SECTION; char *aarg0, *aarg1, *aarg2, *aarg3, *aarg4; char *aarg5, *aarg6, *aarg7, *aarg8, *aarg9; char *aarg10, *aarg11, *aarg12, *aarg13, *aarg14; EXEC SQL END DECLARE SECTION; static int afterc; /* structures to save entire sets of incremental changes */ struct save_queue *incremental_sq = NULL; struct save_queue *incremental_exec = NULL; struct iupdate { char *table; int beforec; char **before; int afterc; char **after; char *service; }; incremental_init() { int i; for (i = 0; i < MAXARGC; i++) { before[i] = &beforeb[i][0]; after[i] = &afterb[i][0]; } barg0 = before[0]; barg1 = before[1]; barg2 = before[2]; barg3 = before[3]; barg4 = before[4]; barg5 = before[5]; barg6 = before[6]; barg7 = before[7]; barg8 = before[8]; barg9 = before[9]; barg10 = before[10]; barg11 = before[11]; barg12 = before[12]; barg13 = before[13]; barg14 = before[14]; aarg0 = after[0]; aarg1 = after[1]; aarg2 = after[2]; aarg3 = after[3]; aarg4 = after[4]; aarg5 = after[5]; aarg6 = after[6]; aarg7 = after[7]; aarg8 = after[8]; aarg9 = after[9]; aarg10 = after[10]; aarg11 = after[11]; aarg12 = after[12]; aarg13 = after[13]; aarg14 = after[14]; if (incremental_sq == NULL) incremental_sq = sq_create(); if (incremental_exec == NULL) incremental_exec = sq_create(); } incremental_before(table, qual, argv) char *table; EXEC SQL BEGIN DECLARE SECTION; char *qual; EXEC SQL END DECLARE SECTION; char **argv; { EXEC SQL BEGIN DECLARE SECTION; int id; EXEC SQL END DECLARE SECTION; char buffer[512], *name; beforetable = table; if (!strcmp(table, "users")) { EXEC SQL SELECT login, CHAR(uid), shell, last, first, middle, CHAR(status), clearid, type INTO :barg0, :barg1, :barg2, :barg3, :barg4, :barg5, :barg6, :barg7, :barg8 FROM users WHERE :qual; beforec = 9; } else if (!strcmp(table, "machine")) { EXEC SQL SELECT name, type INTO :barg0, :barg1 FROM machine WHERE :qual; beforec = 2; } else if (!strcmp(table, "cluster")) { EXEC SQL SELECT name, description, location INTO :barg0, :barg1, :barg2 FROM cluster WHERE :qual; beforec = 3; } else if (!strcmp(table, "mcmap")) { strcpy(barg0, argv[0]); strcpy(barg1, argv[1]); beforec = 2; } else if (!strcmp(table, "svc")) { strcpy(barg0, argv[0]); strcpy(barg1, argv[1]); strcpy(barg2, argv[2]); beforec = 3; } else if (!strcmp(table, "filesys")) { EXEC SQL SELECT label, type, CHAR(mach_id), name, mount, access, comments, CHAR(owner), CHAR(owners), CHAR(createflg), lockertype INTO :barg0, :barg1, :barg2, :barg3, :barg4, :barg5, :barg6, :barg7, :barg8, :barg9, :barg10 FROM filesys WHERE :qual; name = malloc(0); id = atoi(barg2); id_to_name(id, "MACHINE", &name); strcpy(barg2, name); id = atoi(barg7); id_to_name(id, "USER", &name); strcpy(barg7, name); id = atoi(barg8); id_to_name(id, "LIST", &name); strcpy(barg8, name); free(name); beforec = 11; } else if (!strcmp(table, "quota")) { strcpy(barg0, "?"); strcpy(barg1, argv[1]); strcpy(barg2, "?"); sprintf(buffer, "%s and filesys.filsys_id = quota.filsys_id", qual); qual = buffer; EXEC SQL SELECT CHAR(quota.quota), filesys.name INTO :barg3, :barg4 FROM quota, filesys WHERE :qual; beforec = 5; } else if (!strcmp(table, "list")) { EXEC SQL SELECT name, CHAR(active), CHAR(publicflg), CHAR(hidden), CHAR(maillist), CHAR(grouplist), CHAR(gid), acl_type, CHAR(acl_id), description INTO :barg0, :barg1, :barg2, :barg3, :barg4, :barg5, :barg6, :barg7, :barg8, :barg9 FROM list WHERE :qual; beforec = 10; } else if (!strcmp(table, "members")) { id = (int) argv[0]; EXEC SQL SELECT CHAR(grouplist) INTO :barg3 FROM list WHERE list_id = :id; name = malloc(0); id_to_name(id, "LIST", &name); strcpy(barg0, name); strcpy(barg1, argv[1]); id = (int) argv[2]; if (!strcmp(barg1, "USER")) { id_to_name(id, barg1, &name); } else if (!strcmp(barg1, "LIST")) { id_to_name(id, barg1, &name); } else if (!strcmp(barg1, "STRING")) { id_to_name(id, barg1, &name); } else if (!strcmp(barg1, "KERBEROS")) { id_to_name(id, "STRING", &name); } strcpy(barg2, name); free(name); beforec = 4; } /* else com_err(whoami, 0, "unknown table in incremental_before"); */ } incremental_clear_before() { beforec = 0; } incremental_clear_after() { incremental_after("clear", 0); } incremental_after(table, qual, argv) char *table; EXEC SQL BEGIN DECLARE SECTION; char *qual; EXEC SQL END DECLARE SECTION; char **argv; { char buffer[2048], *name; EXEC SQL BEGIN DECLARE SECTION; int id, i; EXEC SQL END DECLARE SECTION; struct iupdate *iu; char **copy_argv(); if (!strcmp(table, "users")) { EXEC SQL SELECT login, CHAR(uid), shell, last, first, middle, CHAR(status), clearid, type INTO :aarg0, :aarg1, :aarg2, :aarg3, :aarg4, :aarg5, :aarg6, :aarg7, :aarg8 FROM users WHERE :qual; afterc = 9; } else if (!strcmp(table, "machine")) { EXEC SQL SELECT name, type INTO :aarg0, :aarg1 FROM machine WHERE :qual; afterc = 2; } else if (!strcmp(table, "cluster")) { EXEC SQL SELECT name, description, location INTO :aarg0, :aarg1, :aarg2 FROM cluster WHERE :qual; afterc = 3; } else if (!strcmp(table, "mcmap")) { strcpy(aarg0, argv[0]); strcpy(aarg1, argv[1]); afterc = 2; } else if (!strcmp(table, "svc")) { strcpy(aarg0, argv[0]); strcpy(aarg1, argv[1]); strcpy(aarg2, argv[2]); afterc = 3; } else if (!strcmp(table, "filesys")) { EXEC SQL SELECT CHAR(label), type, CHAR(mach_id), name, mount, access, comments, CHAR(owner), CHAR(owners), CHAR(createflg), lockertype INTO :aarg0, :aarg1, :aarg2, :aarg3, :aarg4, :aarg5, :aarg6, :aarg7, :aarg8, :aarg9, :aarg10 FROM filesys fs WHERE :qual; name = malloc(0); id = atoi(aarg2); id_to_name(id, "MACHINE", &name); strcpy(aarg2, name); id = atoi(aarg7); id_to_name(id, "USER", &name); strcpy(aarg7, name); id = atoi(aarg8); id_to_name(id, "LIST", &name); strcpy(aarg8, name); free(name); afterc = 11; } else if (!strcmp(table, "quota")) { strcpy(aarg0, "?"); strcpy(aarg1, argv[1]); strcpy(aarg2, "?"); sprintf(buffer, "%s and filesys.filsys_id = quota.filsys_id", qual); qual = buffer; EXEC SQL SELECT CHAR(quota.quota), filesys.name INTO :aarg3, :aarg4 FROM quota, filesys WHERE :qual; afterc = 5; } else if (!strcmp(table, "list")) { EXEC SQL SELECT name, CHAR(active), CHAR(publicflg), CHAR(hidden), CHAR(maillist), CHAR(grouplist), CHAR(gid), acl_type, CHAR(acl_id), description INTO :aarg0, :aarg1, :aarg2, :aarg3, :aarg4, :aarg5, :aarg6, :aarg7, :aarg8, :aarg9 FROM list WHERE :qual; afterc = 10; } else if (!strcmp(table, "members")) { id = (int) argv[0]; EXEC SQL SELECT CHAR(grouplist) INTO :aarg3 FROM list WHERE list_id = :id; name = malloc(0); id_to_name(id, "LIST", &name); strcpy(aarg0, name); strcpy(aarg1, argv[1]); id = (int) argv[2]; if (!strcmp(aarg1, "USER")) { id_to_name(id, aarg1, &name); } else if (!strcmp(aarg1, "LIST")) { id_to_name(id, aarg1, &name); } else if (!strcmp(aarg1, "STRING")) { id_to_name(id, aarg1, &name); } else if (!strcmp(aarg1, "KERBEROS")) { id_to_name(id, "STRING", &name); } strcpy(aarg2, name); free(name); afterc = 4; } else if (!strcmp(table, "clear")) { afterc = 0; table = beforetable; } /* else com_err(whoami, 0, "unknown table in incremental_after"); */ iu = (struct iupdate *) malloc(sizeof(struct iupdate)); iu->table = strsave(table); iu->beforec = beforec; iu->before = copy_argv(before, beforec); iu->afterc = afterc; iu->after = copy_argv(after, afterc); sq_save_data(incremental_sq, iu); #ifdef DEBUG sprintf(buffer, "INCREMENTAL(%s, [", table); for (i = 0; i < beforec; i++) { if (i == 0) strcat(buffer, strtrim(before[0])); else { strcat(buffer, ", "); strcat(buffer, strtrim(before[i])); } } strcat(buffer, "], ["); for (i = 0; i < afterc; i++) { if (i == 0) strcat(buffer, strtrim(after[0])); else { strcat(buffer, ", "); strcat(buffer, strtrim(after[i])); } } strcat(buffer, "])"); com_err(whoami, 0, buffer); #endif DEBUG } /* Called when the current transaction is committed to start any queued * incremental updates. This caches the update table the first time it * is called. */ struct inc_cache { struct inc_cache *next; char *table; char *service; }; incremental_update() { static int inited = 0; static struct inc_cache *cache; struct inc_cache *c; EXEC SQL BEGIN DECLARE SECTION; char tab[17], serv[17]; EXEC SQL END DECLARE SECTION; struct iupdate *iu; if (!inited) { inited++; EXEC SQL DECLARE inc CURSOR FOR SELECT tablename, service FROM incremental; EXEC SQL OPEN inc; while (1) { EXEC SQL FETCH inc INTO :tab, :serv; if (sqlca.sqlcode != 0) break; c = (struct inc_cache *)malloc(sizeof(struct inc_cache)); c->next = cache; c->table = strsave(strtrim(tab)); c->service = strsave(strtrim(serv)); cache = c; } EXEC SQL CLOSE inc; } while (sq_remove_data(incremental_sq, &iu)) { for (c = cache; c; c = c->next) { if (!strcmp(c->table, iu->table)) { iu->service = c->service; sq_save_data(incremental_exec, iu); } } } if (inc_running == 0) next_incremental(); } next_incremental() { struct iupdate *iu; char *argv[MAXARGC * 2 + 4], cafter[3], cbefore[3], prog[BUFSIZ]; int i; if (incremental_exec == NULL) incremental_init(); if (sq_empty(incremental_exec) || (inc_running && now - inc_started < INC_TIMEOUT)) return; if (inc_running) com_err(whoami, 0, "incremental timeout on pid %d", inc_pid); sq_remove_data(incremental_exec, &iu); argv[1] = iu->table; sprintf(cbefore, "%d", iu->beforec); argv[2] = cbefore; sprintf(cafter, "%d", iu->afterc); argv[3] = cafter; for (i = 0; i < iu->beforec; i++) argv[4 + i] = iu->before[i]; for (i = 0; i < iu->afterc; i++) argv[4 + iu->beforec + i] = iu->after[i]; sprintf(prog, "%s/%s.incr", BIN_DIR, iu->service); #ifdef DEBUG com_err(whoami, 0, "forking %s", prog); #endif argv[0] = prog; argv[4 + iu->beforec + iu->afterc] = 0; inc_pid = vfork(); switch (inc_pid) { case 0: execv(prog, argv); _exit(1); case -1: com_err(whoami, 0, "Failed to start incremental update"); break; default: inc_running = 1; inc_started = now; } free_argv(iu->before, iu->beforec); free_argv(iu->after, iu->afterc); free(iu->table); free(iu); } /* Called when the current transaction is aborted to throw away any queued * incremental updates */ incremental_flush() { struct iupdate *iu; while (sq_get_data(incremental_sq, &iu)) { free_argv(iu->before, iu->beforec); free_argv(iu->after, iu->afterc); free(iu->table); free(iu); } sq_destroy(incremental_sq); incremental_sq = sq_create(); } char **copy_argv(argv, argc) char **argv; int argc; { char **ret = (char **)malloc(sizeof(char *) * argc); while (--argc >= 0) ret[argc] = strsave(strtrim(argv[argc])); return(ret); } free_argv(argv, argc) char **argv; int argc; { while (--argc >= 0) free(argv[argc]); free(argv); }