driver.c

开源备份软件源码 AMANDA, the Advanced Maryland Automatic Network Disk Archiver, is a backup system that a

    taper_tape_error = NULL;
    taper_disk = NULL;
    taper_ev_read = NULL;

    schedule_done = nodump;
    force_flush = 0;

    if(!need_degraded) startaflush();

    if(!nodump)
	schedule_ev_read = event_register((event_id_t)0, EV_READFD, read_schedule, NULL);

    short_dump_state();
    event_loop(0);

    force_flush = 1;

    /* mv runq to directq */
    while (!empty(runq)) {
	diskp = dequeue_disk(&runq);
	headqueue_disk(&directq, diskp);
    }

    /* handle any remaining dumps by dumping directly to tape, if possible */
    while(!empty(directq) && taper > 0) {
	diskp = dequeue_disk(&directq);
	if (diskp->to_holdingdisk == HOLD_REQUIRED) {
	    log_add(L_FAIL, _("%s %s %s %d [%s]"),
		diskp->host->hostname, diskp->name, sched(diskp)->datestamp,
		sched(diskp)->level,
		_("can't dump required holdingdisk"));
	}
	else if (!degraded_mode) {
	    taper_state |= TAPER_STATE_DUMP_TO_TAPE;
	    dump_to_tape(diskp);
	    event_loop(0);
	    taper_state &= !TAPER_STATE_DUMP_TO_TAPE;
	}
	else
	    log_add(L_FAIL, _("%s %s %s %d [%s]"),
		diskp->host->hostname, diskp->name, sched(diskp)->datestamp,
		sched(diskp)->level,
		diskp->to_holdingdisk == HOLD_AUTO ?
		    _("no more holding disk space") :
		    _("can't dump no-hold disk in degraded mode"));
    }

    /* fill up the tape or start new one for taperflush */
    startaflush();
    event_loop(0);

    short_dump_state();				/* for amstatus */

    g_printf(_("driver: QUITTING time %s telling children to quit\n"),
           walltime_str(curclock()));
    fflush(stdout);

    if(!nodump) {
	for(dumper = dmptable; dumper < dmptable + inparallel; dumper++) {
	    if(dumper->fd >= 0)
		dumper_cmd(dumper, QUIT, NULL);
	}
    }

    if(taper >= 0) {
	taper_cmd(QUIT, NULL, NULL, 0, NULL);
    }

    /* wait for all to die */
    wait_children(600);

    /* cleanup */
    holding_cleanup(NULL, NULL);

    amfree(newdir);

    check_unfree_serial();
    g_printf(_("driver: FINISHED time %s\n"), walltime_str(curclock()));
    fflush(stdout);
    log_add(L_FINISH,_("date %s time %s"), driver_timestamp, walltime_str(curclock()));
    amfree(driver_timestamp);

    amfree(dumper_program);
    amfree(taper_program);

    dbclose();

    return 0;
}

/* sleep up to count seconds, and wait for terminating child process */
/* if sleep is negative, this function will not timeout              */
/* exit once all child process are finished or the timout expired    */
/* return 0 if no more children to wait                              */
/* return 1 if some children are still alive                         */
static int
wait_children(int count)
{
    pid_t     pid;
    amwait_t  retstat;
    char     *who;
    char     *what;
    int       code=0;
    dumper_t *dumper;
    int       wait_errno;

    do {
	do {
	    pid = waitpid((pid_t)-1, &retstat, WNOHANG);
	    wait_errno = errno;
	    if (pid > 0) {
		what = NULL;
		if (! WIFEXITED(retstat)) {
		    what = _("signal");
		    code = WTERMSIG(retstat);
		} else if (WEXITSTATUS(retstat) != 0) {
		    what = _("code");
		    code = WEXITSTATUS(retstat);
		}
		who = NULL;
		for (dumper = dmptable; dumper < dmptable + inparallel;
		     dumper++) {
		    if (pid == dumper->pid) {
			who = stralloc(dumper->name);
			dumper->pid = -1;
			break;
		    }
		    if (dumper->chunker && pid == dumper->chunker->pid) {
			who = stralloc(dumper->chunker->name);
			dumper->chunker->pid = -1;
			break;
		    }
		}
		if (who == NULL && pid == taper_pid) {
		    who = stralloc("taper");
		    taper_pid = -1;
		}
		if(what != NULL && who == NULL) {
		    who = stralloc("unknown");
		}
		if(who && what) {
		    log_add(L_WARNING, _("%s pid %u exited with %s %d\n"), who, 
			    (unsigned)pid, what, code);
		    g_printf(_("driver: %s pid %u exited with %s %d\n"), who,
			   (unsigned)pid, what, code);
		}
		amfree(who);
	    }
	} while (pid > 0 || wait_errno == EINTR);
	if (errno != ECHILD)
	    sleep(1);
	if (count > 0)
	    count--;
    } while ((errno != ECHILD) && (count != 0));
    return (errno != ECHILD);
}

static void
kill_children(int signal)
{
    dumper_t *dumper;

    if(!nodump) {
        for(dumper = dmptable; dumper < dmptable + inparallel; dumper++) {
	    if (!dumper->down && dumper->pid > 1) {
		g_printf(_("driver: sending signal %d to %s pid %u\n"), signal,
		       dumper->name, (unsigned)dumper->pid);
		if (kill(dumper->pid, signal) == -1 && errno == ESRCH) {
		    if (dumper->chunker)
			dumper->chunker->pid = 0;
		}
		if (dumper->chunker && dumper->chunker->pid > 1) {
		    g_printf(_("driver: sending signal %d to %s pid %u\n"), signal,
			   dumper->chunker->name,
			   (unsigned)dumper->chunker->pid);
		    if (kill(dumper->chunker->pid, signal) == -1 &&
			errno == ESRCH)
			dumper->chunker->pid = 0;
		}
	    }
        }
    }

    if(taper_pid > 1)
	g_printf(_("driver: sending signal %d to %s pid %u\n"), signal,
	       "taper", (unsigned)taper_pid);
	if (kill(taper_pid, signal) == -1 && errno == ESRCH)
	    taper_pid = 0;
}

static void
wait_for_children(void)
{
    dumper_t *dumper;

    if(!nodump) {
	for(dumper = dmptable; dumper < dmptable + inparallel; dumper++) {
	    if (dumper->pid > 1 && dumper->fd >= 0) {
		dumper_cmd(dumper, QUIT, NULL);
		if (dumper->chunker && dumper->chunker->pid > 1 &&
		    dumper->chunker->fd >= 0)
		    chunker_cmd(dumper->chunker, QUIT, NULL);
	    }
	}
    }

    if(taper_pid > 1 && taper > 0) {
	taper_cmd(QUIT, NULL, NULL, 0, NULL);
    }

    if(wait_children(60) == 0)
	return;

    kill_children(SIGHUP);
    if(wait_children(60) == 0)
	return;

    kill_children(SIGKILL);
    if(wait_children(-1) == 0)
	return;

}

static void
startaflush(void)
{
    disk_t *dp = NULL;
    disk_t *fit = NULL;
    char *datestamp;
    int extra_tapes = 0;
    char *qname;
    TapeAction result_tape_action;

    result_tape_action = tape_action();

    if (result_tape_action & TAPE_ACTION_NEW_TAPE) {
	taper_state &= !TAPER_STATE_WAIT_FOR_TAPE;
	taper_cmd(NEW_TAPE, NULL, NULL, 0, NULL);
    } else if (result_tape_action & TAPE_ACTION_NO_NEW_TAPE) {
	taper_state &= !TAPER_STATE_WAIT_FOR_TAPE;
	taper_cmd(NO_NEW_TAPE, NULL, NULL, 0, NULL);
    }

    if (!degraded_mode && !taper_busy && !empty(tapeq) &&
	(result_tape_action & TAPE_ACTION_START_A_FLUSH)) {
	
	datestamp = sched(tapeq.head)->datestamp;
	switch(conf_taperalgo) {
	case ALGO_FIRST:
		dp = dequeue_disk(&tapeq);
		break;
	case ALGO_FIRSTFIT:
		fit = tapeq.head;
		while (fit != NULL) {
		    extra_tapes = (fit->tape_splitsize > (off_t)0) ? 
					conf_runtapes - current_tape : 0;
		    if(sched(fit)->act_size <= (tape_left +
		             tape_length * (off_t)extra_tapes) &&
			     strcmp(sched(fit)->datestamp, datestamp) <= 0) {
			dp = fit;
			fit = NULL;
		    }
		    else {
			fit = fit->next;
		    }
		}
		if(dp) remove_disk(&tapeq, dp);
		break;
	case ALGO_LARGEST:
		fit = dp = tapeq.head;
		while (fit != NULL) {
		    if(sched(fit)->act_size > sched(dp)->act_size &&
		       strcmp(sched(fit)->datestamp, datestamp) <= 0) {
			dp = fit;
		    }
		    fit = fit->next;
		}
		if(dp) remove_disk(&tapeq, dp);
		break;
	case ALGO_LARGESTFIT:
		fit = tapeq.head;
		while (fit != NULL) {
		    extra_tapes = (fit->tape_splitsize > (off_t)0) ? 
					conf_runtapes - current_tape : 0;
		    if(sched(fit)->act_size <=
		       (tape_left + tape_length * (off_t)extra_tapes) &&
		       (!dp || sched(fit)->act_size > sched(dp)->act_size) &&
		       strcmp(sched(fit)->datestamp, datestamp) <= 0) {
			dp = fit;
		    }
		    fit = fit->next;
		}
		if(dp) remove_disk(&tapeq, dp);
		break;
	case ALGO_SMALLEST:
		break;
	case ALGO_LAST:
		dp = tapeq.tail;
		remove_disk(&tapeq, dp);
		break;
	}
	if(!dp) { /* ALGO_SMALLEST, or default if nothing fit. */
	    if(conf_taperalgo != ALGO_SMALLEST)  {
		g_fprintf(stderr,
		   _("driver: startaflush: Using SMALLEST because nothing fit\n"));
	    }
	    fit = dp = tapeq.head;
	    while (fit != NULL) {
		if(sched(fit)->act_size < sched(dp)->act_size &&
		   strcmp(sched(fit)->datestamp, datestamp) <= 0) {
		    dp = fit;
		}
	        fit = fit->next;
	    }
	    if(dp) remove_disk(&tapeq, dp);
	}
	if(taper_ev_read == NULL) {
	    taper_ev_read = event_register((event_id_t)taper, EV_READFD,
					   handle_taper_result, NULL);
	}
	if (dp) {
	    taper_disk = dp;
	    taper_busy = 1;
	    taper_input_error = NULL;
	    taper_tape_error = NULL;
	    taper_result = LAST_TOK;
	    taper_sendresult = 0;
	    taper_first_label = NULL;
	    taper_written = 0;
	    taper_state &= !TAPER_STATE_DUMP_TO_TAPE;
	    taper_dumper = NULL;
	    qname = quote_string(dp->name);
	    taper_cmd(FILE_WRITE, dp, sched(dp)->destname, sched(dp)->level,
		      sched(dp)->datestamp);
	    g_fprintf(stderr,_("driver: startaflush: %s %s %s %lld %lld\n"),
		    taperalgo2str(conf_taperalgo), dp->host->hostname, qname,
		    (long long)sched(taper_disk)->act_size,
		    (long long)tape_left);
	    if(sched(dp)->act_size <= tape_left)
		tape_left -= sched(dp)->act_size;
	    else
		tape_left = (off_t)0;
	    amfree(qname);
	} else {
	    error(_("FATAL: Taper marked busy and no work found."));
	    /*NOTREACHED*/
	}
    } else if(!taper_busy && taper_ev_read != NULL) {
	event_release(taper_ev_read);
	taper_ev_read = NULL;
    }
}

static int
client_constrained(
    disk_t *	dp)
{
    disk_t *dp2;

    /* first, check if host is too busy */

    if(dp->host->inprogress >= dp->host->maxdumps) {
	return 1;
    }

    /* next, check conflict with other dumps on same spindle */

    if(dp->spindle == -1) {	/* but spindle -1 never conflicts by def. */
	return 0;
    }

    for(dp2 = dp->host->disks; dp2 != NULL; dp2 = dp2->hostnext)
	if(dp2->inprogress && dp2->spindle == dp->spindle) {
	    return 1;
	}

    return 0;
}

static void
start_some_dumps(
    disklist_t *	rq)
{
    int cur_idle;
    disk_t *diskp, *delayed_diskp, *diskp_accept;
    assignedhd_t **holdp=NULL, **holdp_accept;
    const time_t now = time(NULL);
    cmd_t cmd;
    int result_argc;
    char *result_argv[MAX_ARGS+1];
    chunker_t *chunker;
    dumper_t *dumper;
    char dumptype;
    char *dumporder;
    int  busy_dumpers = 0;

    idle_reason = IDLE_NO_DUMPERS;
    sleep_time = 0;

    if(dumpers_ev_time != NULL) {
	event_release(dumpers_ev_time);
	dumpers_ev_time = NULL;
    }

    for(dumper = dmptable; dumper < (dmptable+inparallel); dumper++) {
	if( dumper->busy ) {
	    busy_dumpers++;
	}
    }

    for (dumper = dmptable; dumper < dmptable+inparallel; dumper++) {

	if( dumper->busy || dumper->down) {
	    continue;
	}

	if (dumper->ev_read != NULL) {
	    event_release(dumper->ev_read);
	    dumper->ev_read = NULL;
	}

	/*
	 * A potential problem with starting from the bottom of the dump time
	 * distribution is that a slave host will have both one of the shortest
	 * and one of the longest disks, so starting its shortest disk first will
	 * tie up the host and eliminate its longest disk from consideration the
	 * first pass through.  This could cause a big delay in starting that long
	 * disk, which could drag out the whole night's dumps.
	 *
	 * While starting from the top of the dump time distribution solves the
	 * above problem, this turns out to be a bad idea, because the big dumps
	 * will almost certainly pack the holding disk completely, leaving no
	 * room for even one small dump to start.  This ends up shutting out the
	 * small-end dumpers completely (they stay idle).
	 *
	 * The introduction of multiple simultaneous dumps to one host alleviates
	 * the biggest&smallest dumps problem: both can be started at the
	 * beginning.
	 */

	diskp_accept = NULL;
	holdp_accept = NULL;
	delayed_diskp = NULL;

	cur_idle = NOT_IDLE;

	dumporder = getconf_str(CNF_DUMPORDER);
	if(strlen(dumporder) > (size_t)(dumper-dmptable)) {
	    dumptype = dumporder[dumper-dmptable];
	}
	else {
	    if(dumper-dmptable < 3)
		dumptype = 't';
	    else
		dumptype = 'T';