mmstatd.c

mmstatd包含一个 C库和服务器

		for (knod = (struct key_node *)key_list->bottom; knod;
			knod = (struct key_node *)knod->nod.prev)
		    if (knod->hash == hash) break;
	    }
	    kdirection = !kdirection;
	    if (knod) {
		/* Make sure that uid matches entry creator's. Of course uid 0
		 * has total access.
		 */
		if (knod->data->entry.uid != entry->uid &&
			entry->uid != 0) {
		    syslog(LOG_NOTICE, "Unauthorized uid!");
		    ok = FALSE;
		}
		if (ok) {
		    if (type == STAT_DELETE) {
			/* Delete entry */
			UNLINKNODE(data_list, (node *)knod->data);
			freenode((node *)knod->data);
			UNLINKNODE(key_list, (node *)knod);
			freenode((node *)knod);
			return (TRUE);
		    }
		}
	    } else {
		/* Key does not exist */
		if (type == STAT_DELETE) return (TRUE);
		if (type == STAT_UPDATE || type == STAT_RESET) {
		    if (!(entry->autoflush && (type == STAT_UPDATE ?
				    entry->un.update.modifier :
				    entry->un.reset.value) == 0)) {
			register struct data_node *dnod;

			/* Create new entry */
			if ((dnod = (struct data_node *)allocnode(data_list,
					FALSE))) {
			    if ((knod = (struct key_node *)allocnode(key_list,
					    FALSE))) {
				mm_strncpy(dnod->entry.key, entry->key,
					KEY_SIZE - 1);
				dnod->entry.value = type == STAT_UPDATE ?
				    entry->un.update.modifier :
				    entry->un.reset.value;
				dnod->entry.uid = entry->uid;
				dnod->entry.created = dnod->entry.modified =
				    time(NULL);
				dnod->entry.persistant = entry->persistant;
				APPENDNODE(data_list, (node *)dnod);
				knod->hash = hash;
				knod->data = dnod;
				APPENDNODE(key_list, (node *)knod);
			    } else
				dnod = (struct data_node *)freenode(
					(node *)dnod);
			}
			if (!dnod || !knod)
			    return (FALSE);
		    }
		    return (ok);
		}
	    }

	    if (ok) {
		if (type == STAT_UPDATE)
		    knod->data->entry.value += entry->un.update.modifier;
		else if(type == STAT_RESET)
		    knod->data->entry.value = entry->un.reset.value;
		if (knod->data->entry.value == 0 && entry->autoflush) {
		    /* autoflush entry which reached zero, delete it */
		    UNLINKNODE(data_list, (node *)knod->data);
		    freenode((node *)knod->data);
		    UNLINKNODE(key_list, (node *)knod);
		    freenode((node *)knod);
		} else
		    knod->data->entry.modified = time(NULL);
	    }

	} else {
	    register struct key_node *knod, *tknod;
	    register struct data_node *dnod;

	    /* Perform operation on all matching keys which belong to us
	     * in an atomic manner.
	     */
	    knod = (struct key_node *)key_list->top;
	    while (knod) {
		tknod = (struct key_node *)knod->nod.next;
		dnod = knod->data;
		if (dnod->entry.uid == entry->uid &&
			log_match(dnod->entry.key, entry->key)) {
		    if (entry->type == STAT_RESET)
			dnod->entry.value = entry->un.reset.value;
		    else if (type == STAT_UPDATE)
			dnod->entry.value += entry->un.update.modifier;
		    if ((dnod->entry.value == 0 && entry->autoflush) ||
			    type == STAT_DELETE) {
			UNLINKNODE(data_list, (node *)dnod);
			freenode((node *)dnod);
			UNLINKNODE(key_list, (node *)knod);
			freenode((node *)knod);
		    }
		}
		knod = tknod;
	    }

	}

    }

    return (ok);
}


static bool
processlogentries(struct log_entry *entries, int len, bool tmp)
{
    int i;
    bool ok = TRUE;

    for (i = 0; i < len; i++) {
	if (!processlogentry(&entries[i], tmp)) {
	    ok = FALSE;
	    break;
	}
    }

    return (ok);
}


/* Only used for debugging when testing/developing */
/*
static void
debuglogentry(char c, struct log_entry *entry)
{
    switch (entry->type) {
	case STAT_TRANSACT:
	    syslog(LOG_NOTICE, "%c STAT_TRANSACT u=%d p=%d a=%d beg=%d k=%s",
		    c, entry->uid, entry->persistant, entry->autoflush,
		    entry->un.transact.begin, entry->key);
	    break;
	case STAT_NEWFILE:
	    syslog(LOG_NOTICE, "%c STAT_NEWFILE u=%d p=%d a=%d num=%ld k=%s",
		    c, entry->uid, entry->persistant, entry->autoflush,
		    entry->un.newfile.lognum, entry->key);
	    break;
	case STAT_UPDATE:
	    syslog(LOG_NOTICE, "%c STAT_UPDATE u=%d p=%d a=%d mod=%lld k=%s",
		    c, entry->uid, entry->persistant, entry->autoflush,
		    entry->un.update.modifier, entry->key);
	    break;
	case STAT_RESET:
	    syslog(LOG_NOTICE, "%c STAT_RESET u=%d p=%d a=%d val=%lld k=%s",
		    c, entry->uid, entry->persistant, entry->autoflush,
		    entry->un.reset.value, entry->key);
	    break;
	case STAT_DELETE:
	    syslog(LOG_NOTICE, "%c STAT_DELETE u=%d p=%d a=%d k=%s",
		    c, entry->uid, entry->persistant, entry->autoflush,
		    entry->key);
	    break;
	default:
	    syslog(LOG_NOTICE, "%c Unknown entry type!", c);
    }
}
*/


/* This function prepares the db lists, and attempts to load previously saved
 * database and sync state. If part of the database cannot be loaded, we
 * log an error via syslog but will either provide an empty or incomplete
 * db in memory.
 */
/* XXX Make file format portable among various endian architectures */
static void
load_db(long *lognum, off_t *logpos)
{
    char filename[256];
    FILE *fh;
    bool ok;
    u_int32_t version, ver;

    ok = TRUE;
    *lognum = 0;
    *logpos = 0;

    snprintf(filename, 255, "%s/mmstatd.db", CONF.ENV_DIR);
    if ((data_list = openlist(malloc, free, sizeof(struct data_node), 16384,
		    0))) {
	if ((key_list = openlist(malloc, free, sizeof(struct key_node), 16384,
			0))) {
	    if ((fh = fopen(filename, "rb"))) {
		/* We now remember how to load old mmstatd database files,
		 * since at times the format changes accross versions.
		 * so first determine which version, and call the appropriate
		 * function.
		 */
		if (fread(&ver, sizeof(u_int32_t), 1, fh) == 1) {
		    version = 0xFFFFFFFF - ver;
		    switch (version) {
			case 2:
			    /* mmstatd 0.0.2, db v2 */
			    syslog(LOG_NOTICE,
				    "load_db() - loading db of v2");
			    ok = load_db_v0_0_2(fh, lognum, logpos);
			    break;
			case 3:
			    /* mmstatd 0.0.3 db v3 */
			    syslog(LOG_NOTICE,
				    "load_db() - loading db of v3");
			    ok = load_db_v0_0_3(fh, lognum, logpos);
			    break;
			default:
			    {
				/* First version */
				long o_lognum, o_logpos;

				/* Seek back to start since there was no
				 * version info back then
				 */
				fseek(fh, 0, SEEK_SET);
				syslog(LOG_NOTICE,
				       	"load_db() - loading db of v1");
				ok = load_db_v0_0_1(fh, &o_lognum, &o_logpos);
				*lognum = o_lognum;
				*logpos = (off_t)o_logpos;
			    }
			    break;
		    }
		} else ok = FALSE;
		fclose(fh);
	    } else
		syslog(LOG_NOTICE, "load_db() - New database");
	}
    }
    if (!ok) {
	syslog(LOG_NOTICE, "load_db() - Error loading database (corrupt)");
    }
}


static bool
load_db_v0_0_1(FILE *fh, long *lognum, long *logpos)
{
    struct key_node *knod = NULL;
    struct data_node *dnod = NULL;
    u_int64_t hash;
    unsigned char len;
    bool ok = TRUE;

    if (fread(lognum, sizeof(long), 1, fh) != 1 ||
	    fread(logpos, sizeof(long), 1, fh) != 1)
	ok = FALSE;
    while (ok) {
	if (fread(&hash, sizeof(u_int64_t), 1, fh) == 1) {
	    if ((dnod = (struct data_node *)allocnode(data_list, FALSE))) {
		dnod->entry.persistant = TRUE;
		if ((knod = (struct key_node *)allocnode(key_list, FALSE))) {
		    knod->hash = hash;
		    knod->data = dnod;
		    if (fread(&dnod->entry.value, sizeof(int64_t), 1, fh) != 1
			    || fread(&dnod->entry.created, sizeof(time_t), 1,
			       	fh) != 1
			    || fread(&dnod->entry.modified, sizeof(time_t), 1,
			       	fh) != 1
			    || fread(&dnod->entry.uid, sizeof(uid_t), 1,
			       	fh) != 1
			    || fread(&len, sizeof(unsigned char), 1, fh) != 1
			    || fread(&dnod->entry.key, len + 1, 1, fh) != 1)
			ok = FALSE;
		    else {
			APPENDNODE(data_list, (node *)dnod);
			APPENDNODE(key_list, (node *)knod);
		    }
		} else
		    ok = FALSE;
	    } else
		ok = FALSE;
	    if (!ok) {
		if (dnod) freenode((node *)dnod);
		if (knod) freenode((node *)knod);
	    }
	} else
	    break;
    }

    return (ok);
}


static bool
load_db_v0_0_2(FILE *fh, long *lognum, off_t *logpos)
{
    struct key_node *knod = NULL;
    struct data_node *dnod = NULL;
    u_int64_t hash;
    unsigned char len;
    bool ok = TRUE;

    if (fread(lognum, sizeof(long), 1, fh) != 1 ||
	    fread(logpos, sizeof(off_t), 1, fh) != 1)
	ok = FALSE;
    while (ok) {
	if (fread(&hash, sizeof(u_int64_t), 1, fh) == 1) {
	    if ((dnod = (struct data_node *)allocnode(data_list, FALSE))) {
		dnod->entry.persistant = TRUE;
		if ((knod = (struct key_node *)allocnode(key_list, FALSE))) {
		    knod->hash = hash;
		    knod->data = dnod;
		    if (fread(&dnod->entry.value, sizeof(int64_t), 1, fh) != 1
			    || fread(&dnod->entry.created, sizeof(time_t), 1,
			       	fh) != 1
			    || fread(&dnod->entry.modified, sizeof(time_t), 1,
			       	fh) != 1
			    || fread(&dnod->entry.uid, sizeof(uid_t), 1,
			       	fh) != 1
			    || fread(&len, sizeof(unsigned char), 1, fh) != 1
			    || fread(&dnod->entry.key, len + 1, 1, fh) != 1)
			ok = FALSE;
		    else {
			APPENDNODE(data_list, (node *)dnod);
			APPENDNODE(key_list, (node *)knod);
		    }
		} else
		    ok = FALSE;
	    } else
		ok = FALSE;
	    if (!ok) {
		if (dnod) freenode((node *)dnod);
		if (knod) freenode((node *)knod);
	    }
	} else
	    break;
    }

    return (ok);
}


static bool
load_db_v0_0_3(FILE *fh, long *lognum, off_t *logpos)
{
    struct key_node *knod = NULL;
    struct data_node *dnod = NULL;
    u_int64_t hash;
    size_t len;
    bool ok = TRUE;

    if (fread(lognum, sizeof(long), 1, fh) != 1 ||
	    fread(logpos, sizeof(off_t), 1, fh) != 1)
	ok = FALSE;
    while (ok) {
	if (fread(&hash, sizeof(u_int64_t), 1, fh) == 1) {
	    if ((dnod = (struct data_node *)allocnode(data_list, FALSE))) {
		dnod->entry.persistant = TRUE;
		if ((knod = (struct key_node *)allocnode(key_list, FALSE))) {
		    knod->hash = hash;
		    knod->data = dnod;
		    if (fread(&dnod->entry.value, sizeof(int64_t), 1, fh) != 1
			    || fread(&dnod->entry.created, sizeof(time_t), 1,
			       	fh) != 1
			    || fread(&dnod->entry.modified, sizeof(time_t), 1,
			       	fh) != 1
			    || fread(&dnod->entry.uid, sizeof(uid_t), 1,
			       	fh) != 1
			    || fread(&len, sizeof(size_t), 1, fh) != 1
			    || fread(&dnod->entry.key, len + 1, 1, fh) != 1)
			ok = FALSE;
		    else {
			APPENDNODE(data_list, (node *)dnod);
			APPENDNODE(key_list, (node *)knod);
		    }
		} else
		    ok = FALSE;
	    } else
		ok = FALSE;
	    if (!ok) {
		if (dnod) freenode((node *)dnod);
		if (knod) freenode((node *)knod);
	    }
	} else
	    break;
    }

    return (ok);
}


/* This function syncs the memory db to disk with current sync info.
 * If save was successful, obsolete recovery logs are deleted.
 * We warn through syslog if the sync could not be performed.
 * If all is TRUE, all logs are deleted after sync.
 */
/* XXX Make file format portable among various endian architectures */
static void
sync_db(long lognum, off_t logpos, bool all)
{
    char old_db[256], new_db[256];
    FILE *fh;
    DIR *dir;
    struct key_node *knod;
    size_t len;
    u_int32_t version = 0xFFFFFFFF - 3;
    bool ok;

    /* We use a technique which permits to retrieve the previous state of
     * the db in case we could not save properly, using rename().
     */
    ok = TRUE;

    snprintf(old_db, 255, "%s/mmstatd.db", CONF.ENV_DIR);
    snprintf(new_db, 255, "%s/mmstatd.tdb", CONF.ENV_DIR);

    /* Use stdio buffering since we are about to perform many small writes */
    if ((fh = fopen(new_db, "wb"))) {
	fchmod(fileno(fh), 0600);
	/* Write db version and current sync state */
	if (fwrite(&version, sizeof(u_int32_t), 1, fh) == 1 &&
		fwrite(&lognum, sizeof(long), 1, fh) == 1 &&
		fwrite(&logpos, sizeof(off_t), 1, fh) == 1) {
	    /* DB contents */
	    for (knod = (struct key_node *)key_list->top; knod && ok;
		    knod = (struct key_node *)knod->nod.next) {
		len = mm_strlen(knod->data->entry.key);
		if (knod->data->entry.persistant) {
		    if (fwrite(&knod->hash, sizeof(u_int64_t), 1, fh)
			    != 1 ||
			    fwrite(&knod->data->entry.value,
				sizeof(int64_t), 1, fh) != 1 ||
			    fwrite(&knod->data->entry.created, sizeof(time_t),
				1, fh) != 1 ||
			    fwrite(&knod->data->entry.modified, sizeof(time_t),
				1, fh) != 1 ||
			    fwrite(&knod->data->entry.uid, sizeof(uid_t), 1,
				fh) != 1 ||
			    fwrite(&len, sizeof(size_t), 1, fh) != 1 ||
			    fwrite(knod->data->entry.key, len + 1, 1, fh)
			    != 1) {
			ok = FALSE;
			break;
		    }
		}
	    }
	    kdirection = TRUE;
	} else
	    ok = FALSE;
	fflush(fh);
	fclose(fh);
    } else
	ok = FALSE;

    if (ok)
	if (rename(new_db, old_db) == -1) ok = FALSE;

    if (!ok) unlink(new_db);
    else {
	/* Scan for recovery logs and unlink obsolete ones */
	if ((dir = opendir(CONF.ENV_DIR))) {
	    char *name, filename[256];
	    struct dirent *dire;
	    long i;

	    while ((dire = readdir(dir))) {
		name = dire->d_name;
		if (log_match(name, "????????.log")) {
		    i = atol(name);
		    if (all || i < lognum) {
			snprintf(filename, 255, "%s/%s", CONF.ENV_DIR, name);
			unlink(filename);
		    }
		}
	    }
	    closedir(dir);
	}
    }
}


static void
free_db(void)
{
    if (data_list) data_list = closelist(data_list);
    if (key_list) key_list = closelist(key_list);
}


/* This function loads the db, attempts to perform recovery processing the
 * logs, resyncs the db and unloads everything. This function is intended to
 * be executed when the logging daemon process is not running.
 */
static void
recover_db(void)
{
    int fd;
    long lognum, len;
    off_t logpos;
    char filename[256];
    struct log_entry lentry[MAX_TRANSACT + 1];
    bool ok;

    syslog(LOG_NOTICE, "Recovering last db modifications from logs");

    /* First obtain our last position in the last logfile so that we do not
     * process logs which have already been applied to the db before last sync.
     */
    ok = FALSE;
    snprintf(filename, 255, "%s/%s", CONF.ENV_DIR, "mmstatd.db");
    load_db(&lognum, &logpos);

    /* If any, start processing logs, but make sure to only start after
     * lognum/logpos, if there are any remaining, since we are only
     * performing recovery of unsynced pending logs.
     */
    snprintf(filename, 255, "%s/%08ld.log", CONF.ENV_DIR, lognum);
    if ((fd = open(filename, O_RDONLY)) != -1) {
	if (logpos > 0) lseek(fd, logpos, SEEK_SET);
	while ((len = readlogentries(-1, lentry, MAX_TRANSACT, &fd, &lognum,
			&logpos)))
	    if (!processlogentries(lentry, len, FALSE)) {
		syslog(LOG_NOTICE, "recover_db() - processlogentries()");
		break;
	    }
	close(fd);
    } else
	syslog(LOG_NOTICE, "recover_db() - open(%s)", filename);

    /* Sync back db to disk and delete obsolete recovery logs */
    lognum = 0;
    logpos = 0;
    sync_db(lognum, logpos, TRUE);
    free_db();
}


/* key char array should be KEY_SIZE bytes in size */
static void
writestats(int fd, char *key)
{
    u_int64_t hash;
    struct data_node *dnod;
    struct key_node *knod;
    char *ptr;

    /* Sanity check */