driver.c

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

	}

	for(diskp = rq->head; diskp != NULL; diskp = diskp->next) {
	    assert(diskp->host != NULL && sched(diskp) != NULL);

	    if (diskp->host->start_t > now) {
		cur_idle = max(cur_idle, IDLE_START_WAIT);
		if (delayed_diskp == NULL || sleep_time > diskp->host->start_t) {
		    delayed_diskp = diskp;
		    sleep_time = diskp->host->start_t;
		}
	    } else if(diskp->start_t > now) {
		cur_idle = max(cur_idle, IDLE_START_WAIT);
		if (delayed_diskp == NULL || sleep_time > diskp->start_t) {
		    delayed_diskp = diskp;
		    sleep_time = diskp->start_t;
		}
	    } else if (diskp->host->netif->curusage > 0 &&
		       sched(diskp)->est_kps > free_kps(diskp->host->netif)) {
		cur_idle = max(cur_idle, IDLE_NO_BANDWIDTH);
	    } else if(sched(diskp)->no_space) {
		cur_idle = max(cur_idle, IDLE_NO_DISKSPACE);
	    } else if (diskp->to_holdingdisk == HOLD_NEVER) {
		cur_idle = max(cur_idle, IDLE_NO_HOLD);
	    } else if ((holdp =
		find_diskspace(sched(diskp)->est_size, &cur_idle, NULL)) == NULL) {
		cur_idle = max(cur_idle, IDLE_NO_DISKSPACE);
		if (empty(tapeq) && busy_dumpers == 0) {
		    remove_disk(rq, diskp);
		    enqueue_disk(&directq, diskp);
		}
	    } else if (client_constrained(diskp)) {
		free_assignedhd(holdp);
		cur_idle = max(cur_idle, IDLE_CLIENT_CONSTRAINED);
	    } else {

		/* disk fits, dump it */
		int accept = !diskp_accept;
		if(!accept) {
		    switch(dumptype) {
		      case 's': accept = (sched(diskp)->est_size < sched(diskp_accept)->est_size);
				break;
		      case 'S': accept = (sched(diskp)->est_size > sched(diskp_accept)->est_size);
				break;
		      case 't': accept = (sched(diskp)->est_time < sched(diskp_accept)->est_time);
				break;
		      case 'T': accept = (sched(diskp)->est_time > sched(diskp_accept)->est_time);
				break;
		      case 'b': accept = (sched(diskp)->est_kps < sched(diskp_accept)->est_kps);
				break;
		      case 'B': accept = (sched(diskp)->est_kps > sched(diskp_accept)->est_kps);
				break;
		      default:  log_add(L_WARNING, _("Unknown dumporder character \'%c\', using 's'.\n"),
					dumptype);
				accept = (sched(diskp)->est_size < sched(diskp_accept)->est_size);
				break;
		    }
		}
		if(accept) {
		    if( !diskp_accept || !degraded_mode || diskp->priority >= diskp_accept->priority) {
			if(holdp_accept) free_assignedhd(holdp_accept);
			diskp_accept = diskp;
			holdp_accept = holdp;
		    }
		    else {
			free_assignedhd(holdp);
		    }
		}
		else {
		    free_assignedhd(holdp);
		}
	    }
	}

	diskp = diskp_accept;
	holdp = holdp_accept;

	idle_reason = max(idle_reason, cur_idle);

	/*
	 * If we have no disk at this point, and there are disks that
	 * are delayed, then schedule a time event to call this dumper
	 * with the disk with the shortest delay.
	 */
	if (diskp == NULL && delayed_diskp != NULL) {
	    assert(sleep_time > now);
	    sleep_time -= now;
	    dumpers_ev_time = event_register((event_id_t)sleep_time, EV_TIME,
		handle_dumpers_time, &runq);
	    return;
	} else if (diskp != NULL) {
	    sched(diskp)->act_size = (off_t)0;
	    allocate_bandwidth(diskp->host->netif, sched(diskp)->est_kps);
	    sched(diskp)->activehd = assign_holdingdisk(holdp, diskp);
	    amfree(holdp);
	    sched(diskp)->destname = newstralloc(sched(diskp)->destname,
						 sched(diskp)->holdp[0]->destname);
	    diskp->host->inprogress++;	/* host is now busy */
	    diskp->inprogress = 1;
	    sched(diskp)->dumper = dumper;
	    sched(diskp)->timestamp = now;

	    dumper->busy = 1;		/* dumper is now busy */
	    dumper->dp = diskp;		/* link disk to dumper */
	    remove_disk(rq, diskp);		/* take it off the run queue */

	    sched(diskp)->origsize = (off_t)-1;
	    sched(diskp)->dumpsize = (off_t)-1;
	    sched(diskp)->dumptime = (time_t)0;
	    sched(diskp)->tapetime = (time_t)0;
	    chunker = dumper->chunker;
	    chunker->result = LAST_TOK;
	    dumper->result = LAST_TOK;
	    startup_chunk_process(chunker,chunker_program);
	    chunker_cmd(chunker, START, (void *)driver_timestamp);
	    chunker->dumper = dumper;
	    chunker_cmd(chunker, PORT_WRITE, diskp);
	    cmd = getresult(chunker->fd, 1, &result_argc, result_argv, MAX_ARGS+1);
	    if(cmd != PORT) {
		assignedhd_t **h=NULL;
		int activehd;

		g_printf(_("driver: did not get PORT from %s for %s:%s\n"),
		       chunker->name, diskp->host->hostname, diskp->name);
		fflush(stdout);

		deallocate_bandwidth(diskp->host->netif, sched(diskp)->est_kps);
		h = sched(diskp)->holdp;
		activehd = sched(diskp)->activehd;
		h[activehd]->used = 0;
		h[activehd]->disk->allocated_dumpers--;
		adjust_diskspace(diskp, DONE);
		delete_diskspace(diskp);
		diskp->host->inprogress--;
		diskp->inprogress = 0;
		sched(diskp)->dumper = NULL;
		dumper->busy = 0;
		dumper->dp = NULL;
		sched(diskp)->dump_attempted++;
		free_serial_dp(diskp);
		if(sched(diskp)->dump_attempted < 2)
		    enqueue_disk(rq, diskp);
	    }
	    else {
		dumper->ev_read = event_register((event_id_t)dumper->fd, EV_READFD,
						 handle_dumper_result, dumper);
		chunker->ev_read = event_register((event_id_t)chunker->fd, EV_READFD,
						   handle_chunker_result, chunker);
		dumper->output_port = atoi(result_argv[2]);

		dumper_cmd(dumper, PORT_DUMP, diskp);
	    }
	    diskp->host->start_t = now + 15;
	}
    }
}

/*
 * This gets called when a dumper is delayed for some reason.  It may
 * be because a disk has a delayed start, or amanda is constrained
 * by network or disk limits.
 */

static void
handle_dumpers_time(
    void *	cookie)
{
    disklist_t *runq = cookie;
    event_release(dumpers_ev_time);
    dumpers_ev_time = NULL; 
    start_some_dumps(runq);
}

static void
dump_schedule(
    disklist_t *qp,
    char *	str)
{
    disk_t *dp;
    char *qname;

    g_printf(_("dump of driver schedule %s:\n--------\n"), str);

    for(dp = qp->head; dp != NULL; dp = dp->next) {
        qname = quote_string(dp->name);
	g_printf("  %-20s %-25s lv %d t %5lu s %lld p %d\n",
	       dp->host->hostname, qname, sched(dp)->level,
	       sched(dp)->est_time,
	       (long long)sched(dp)->est_size, sched(dp)->priority);
        amfree(qname);
    }
    g_printf("--------\n");
}

static void
start_degraded_mode(
    /*@keep@*/ disklist_t *queuep)
{
    disk_t *dp;
    disklist_t newq;
    off_t est_full_size;
    char *qname;

    newq.head = newq.tail = 0;

    dump_schedule(queuep, _("before start degraded mode"));

    est_full_size = (off_t)0;
    while(!empty(*queuep)) {
	dp = dequeue_disk(queuep);

	qname = quote_string(dp->name);
	if(sched(dp)->level != 0)
	    /* go ahead and do the disk as-is */
	    enqueue_disk(&newq, dp);
	else {
	    if (reserved_space + est_full_size + sched(dp)->est_size
		<= total_disksize) {
		enqueue_disk(&newq, dp);
		est_full_size += sched(dp)->est_size;
	    }
	    else if(sched(dp)->degr_level != -1) {
		sched(dp)->level = sched(dp)->degr_level;
		sched(dp)->dumpdate = sched(dp)->degr_dumpdate;
		sched(dp)->est_nsize = sched(dp)->degr_nsize;
		sched(dp)->est_csize = sched(dp)->degr_csize;
		sched(dp)->est_time = sched(dp)->degr_time;
		sched(dp)->est_kps  = sched(dp)->degr_kps;
		enqueue_disk(&newq, dp);
	    }
	    else {
		log_add(L_FAIL,_("%s %s %s %d [can't switch to incremental dump]"),
		        dp->host->hostname, qname, sched(dp)->datestamp,
			sched(dp)->level);
	    }
	}
        amfree(qname);
    }

    /*@i@*/ *queuep = newq;
    degraded_mode = 1;

    dump_schedule(queuep, _("after start degraded mode"));
}


static void
continue_port_dumps(void)
{
    disk_t *dp, *ndp;
    assignedhd_t **h;
    int active_dumpers=0, busy_dumpers=0, i;
    dumper_t *dumper;

    /* First we try to grant diskspace to some dumps waiting for it. */
    for( dp = roomq.head; dp; dp = ndp ) {
	ndp = dp->next;
	/* find last holdingdisk used by this dump */
	for( i = 0, h = sched(dp)->holdp; h[i+1]; i++ ) {
	    (void)h; /* Quiet lint */
	}
	/* find more space */
	h = find_diskspace( sched(dp)->est_size - sched(dp)->act_size,
			    &active_dumpers, h[i] );
	if( h ) {
	    for(dumper = dmptable; dumper < dmptable + inparallel &&
				   dumper->dp != dp; dumper++) {
		(void)dp; /* Quiet lint */
	    }
	    assert( dumper < dmptable + inparallel );
	    sched(dp)->activehd = assign_holdingdisk( h, dp );
	    chunker_cmd( dumper->chunker, CONTINUE, dp );
	    amfree(h);
	    remove_disk( &roomq, dp );
	}
    }

    /* So for some disks there is less holding diskspace available than
     * was asked for. Possible reasons are
     * a) diskspace has been allocated for other dumps which are
     *    still running or already being written to tape
     * b) all other dumps have been suspended due to lack of diskspace
     * c) this dump doesn't fit on all the holding disks
     * Case a) is not a problem. We just wait for the diskspace to
     * be freed by moving the current disk to a queue.
     * If case b) occurs, we have a deadlock situation. We select
     * a dump from the queue to be aborted and abort it. It will
     * be retried later dumping to disk.
     * If case c) is detected, the dump is aborted. Next time
     * it will be dumped directly to tape. Actually, case c is a special
     * manifestation of case b) where only one dumper is busy.
     */
    for(dp=NULL, dumper = dmptable; dumper < (dmptable+inparallel); dumper++) {
	if( dumper->busy ) {
	    busy_dumpers++;
	    if( !find_disk(&roomq, dumper->dp) ) {
		active_dumpers++;
	    } else if( !dp || 
		       sched(dp)->est_size > sched(dumper->dp)->est_size ) {
		dp = dumper->dp;
	    }
	}
    }
    if((dp != NULL) && (active_dumpers == 0) && (busy_dumpers > 0) && 
        ((!taper_busy && empty(tapeq)) || degraded_mode) &&
	pending_aborts == 0 ) { /* not case a */
	if( busy_dumpers == 1 ) { /* case c */
	    sched(dp)->no_space = 1;
	}
	/* case b */
	/* At this time, dp points to the dump with the smallest est_size.
	 * We abort that dump, hopefully not wasting too much time retrying it.
	 */
	remove_disk( &roomq, dp );
	chunker_cmd( sched(dp)->dumper->chunker, ABORT, NULL);
	dumper_cmd( sched(dp)->dumper, ABORT, NULL );
	pending_aborts++;
    }
}


static void
handle_taper_result(
    void *cookie)
{
    disk_t *dp;
    cmd_t cmd;
    int result_argc;
    char *result_argv[MAX_ARGS+1];
    char *qname;

    (void)cookie;	/* Quiet unused parameter warning */

    assert(cookie == NULL);
    
    do {
        
	short_dump_state();
        
	cmd = getresult(taper, 1, &result_argc, result_argv, MAX_ARGS+1);
        
	switch(cmd) {
            
	case FAILED:	/* FAILED <handle> INPUT-* TAPE-* <input err mesg> <tape err mesg> */
	    if(result_argc != 6) {
		error(_("error: [taper FAILED result_argc != 6: %d"), result_argc);
		/*NOTREACHED*/
	    }
            
	    dp = serial2disk(result_argv[2]);
	    assert(dp == taper_disk);
	    if (!taper_dumper)
		free_serial(result_argv[2]);
            
	    qname = quote_string(dp->name);
	    g_printf(_("driver: finished-cmd time %s taper wrote %s:%s\n"),
		   walltime_str(curclock()), dp->host->hostname, qname);
	    fflush(stdout);
            amfree(qname);

	    if (strcmp(result_argv[3], "INPUT-ERROR") == 0) {
		taper_input_error = stralloc(result_argv[5]);
	    }
	    if (strcmp(result_argv[4], "TAPE-ERROR") == 0) {
		taper_tape_error = stralloc(result_argv[6]);
	    }

	    taper_result = cmd;

	    break;
            
	case PARTIAL:	/* PARTIAL <handle> INPUT-* TAPE-* <stat mess> <input err mesg> <tape err mesg>*/
	case DONE:	/* DONE <handle> INPUT-GOOD TAPE-GOOD <stat mess> <input err mesg> <tape err mesg> */
	    if(result_argc != 7) {
		error(_("error: [taper PARTIAL result_argc != 7: %d"), result_argc);
		/*NOTREACHED*/
	    }
            
	    dp = serial2disk(result_argv[2]);
	    assert(dp == taper_disk);
            if (!taper_dumper)
                free_serial(result_argv[2]);

	    g_printf(_("driver: finished-cmd time %s taper wrote %s:%s\n"),
		   walltime_str(curclock()), dp->host->hostname, dp->name);
	    fflush(stdout);

	    if (strcmp(result_argv[3], "INPUT-ERROR") == 0) {
		taper_input_error = stralloc(result_argv[5]);
	    }
	    if (strcmp(result_argv[4], "TAPE-ERROR") == 0) {
		taper_tape_error = stralloc(result_argv[6]);
	    }

	    taper_result = cmd;

	    break;
            
        case PARTDONE:  /* PARTDONE <handle> <label> <fileno> <stat> */
	    dp = serial2disk(result_argv[2]);
	    assert(dp == taper_disk);
            if (result_argc != 6) {
                error(_("error [taper PARTDONE result_argc != 5: %d]"),
                      result_argc);
		/*NOTREACHED*/
            }
	    if (!taper_first_label) {
		taper_first_label = stralloc(result_argv[3]);
		taper_first_fileno = OFF_T_ATOI(result_argv[4]);
	    }
	    taper_written = OFF_T_ATOI(result_argv[5]);
	    if (taper_written > sched(taper_disk)->act_size)
		sched(taper_disk)->act_size = taper_written;
            
            break;

        case REQUEST_NEW_TAPE:  /* REQUEST-NEW-TAPE */
            if (result_argc != 2) {
                error(_("error [taper REQUEST_NEW_TAPE result_argc != 2: %d]"),
                      result_argc);
		/*NOTREACHED*/
            }
	    taper_state &= !TAPER_STATE_TAPE_STARTED;

	    if (current_tape >= conf_runtapes) {
		taper_cmd(NO_NEW_TAPE, NULL, NULL, 0, NULL);
		log_add(L_WARNING,
			_("Out of tapes; going into degraded mode."));
		start_degraded_mode(&runq);
	    } else {
		TapeAction result_tape_action;

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

	case NEW_TAPE: /* NEW-TAPE <handle> <label> */
            if (result_argc != 3) {
                error(_("error [taper NEW_TAPE result_argc != 3: %d]"),
                      result_argc);
		/*NOTREACHED*/
            }

            /* Update our tape counter and reset tape_left */
	    current_tape++;
	    tape_left = tape_length;
	    taper_state |= TAPER_STATE_TAPE_STARTED;