dumper.c

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

	file->encrypted= 1;
      }
    }
}

/*
 * Send an Amanda dump header to the output file.
 */
static ssize_t
write_tapeheader(
    int		outfd,
    dumpfile_t *file)
{
    char * buffer;
    ssize_t written;

    buffer = build_header(file, DISK_BLOCK_BYTES);

    written = fullwrite(outfd, buffer, DISK_BLOCK_BYTES);
    amfree(buffer);
    if(written == DISK_BLOCK_BYTES)
        return 0;
    if(written < 0)
        return written;

    return -1;
}

static int
do_dump(
    struct databuf *db)
{
    char *indexfile_tmp = NULL;
    char *indexfile_real = NULL;
    char level_str[NUM_STR_SIZE];
    char *fn;
    char *q;
    times_t runtime;
    double dumptime;	/* Time dump took in secs */
    char *errfname = NULL;
    int indexout;
    pid_t indexpid = -1;

    startclock();

    status = 0;
    dump_result = 0;
    dumpbytes = dumpsize = headersize = origsize = (off_t)0;
    fh_init(&file);

    g_snprintf(level_str, SIZEOF(level_str), "%d", level);
    fn = sanitise_filename(diskname);
    errfname = newvstralloc(errfname,
			    AMANDA_TMPDIR,
			    "/", hostname,
			    ".", fn,
			    ".", level_str,
			    ".errout",
			    NULL);
    amfree(fn);
    if((errf = fopen(errfname, "w+")) == NULL) {
	errstr = newvstrallocf(errstr, "errfile open \"%s\": %s",
			      errfname, strerror(errno));
	amfree(errfname);
	goto failed;
    }
    unlink(errfname);				/* so it goes away on close */
    amfree(errfname);

    if (streams[INDEXFD].fd != NULL) {
	indexfile_real = getindexfname(hostname, diskname, dumper_timestamp, level);
	indexfile_tmp = stralloc2(indexfile_real, ".tmp");

	if (mkpdir(indexfile_tmp, 0755, (uid_t)-1, (gid_t)-1) == -1) {
	   errstr = newvstrallocf(errstr,
				 _("err create %s: %s"),
				 indexfile_tmp,
				 strerror(errno));
	   amfree(indexfile_real);
	   amfree(indexfile_tmp);
	   goto failed;
	}
	indexout = open(indexfile_tmp, O_WRONLY | O_CREAT | O_TRUNC, 0600);
	if (indexout == -1) {
	    errstr = newvstrallocf(errstr, _("err open %s: %s"),
			indexfile_tmp, strerror(errno));
	    goto failed;
	} else {
	    if (runcompress(indexout, &indexpid, COMP_BEST) < 0) {
		aclose(indexout);
		goto failed;
	    }
	}
	indexfderror = 0;
	/*
	 * Schedule the indexfd for relaying to the index file
	 */
	security_stream_read(streams[INDEXFD].fd, read_indexfd, &indexout);
    }

    /*
     * We only need to process messages initially.  Once we have done
     * the header, we will start processing data too.
     */
    security_stream_read(streams[MESGFD].fd, read_mesgfd, db);
    set_datafd = 0;

    /*
     * Setup a read timeout
     */
    timeout(conf_dtimeout);

    /*
     * Start the event loop.  This will exit when all three events
     * (read the mesgfd, read the datafd, and timeout) are removed.
     */
    event_loop(0);

    if (!ISSET(status, HEADER_DONE)) {
	dump_result = max(dump_result, 2);
	if (!errstr) errstr = stralloc(_("got no header information"));
    }

    dumpsize -= headersize;		/* don't count the header */
    if (dumpsize <= (off_t)0) {
	dumpsize = (off_t)0;
	dump_result = max(dump_result, 2);
	if (!errstr) errstr = stralloc(_("got no data"));
    }

    if (dump_result > 1)
	goto failed;

    runtime = stopclock();
    dumptime = g_timeval_to_double(runtime);

    amfree(errstr);
    errstr = alloc(128);
    g_snprintf(errstr, 128, _("sec %s kb %lld kps %3.1lf orig-kb %lld"),
	walltime_str(runtime),
	(long long)dumpsize,
	(isnormal(dumptime) ? ((double)dumpsize / (double)dumptime) : 0.0),
	(long long)origsize);
    q = squotef("[%s]", errstr);
    putresult(DONE, _("%s %lld %lld %lu %s\n"), handle,
    		(long long)origsize,
		(long long)dumpsize,
	        (unsigned long)((double)dumptime+0.5), q);
    amfree(q);

    switch(dump_result) {
    case 0:
	log_add(L_SUCCESS, "%s %s %s %d [%s]", hostname, qdiskname, dumper_timestamp, level, errstr);

	break;

    case 1:
	log_start_multiline();
	log_add(L_STRANGE, "%s %s %d [%s]", hostname, qdiskname, level, errstr);
	log_msgout(L_STRANGE);
	log_end_multiline();

	break;
    }

    if (errf) afclose(errf);

    aclose(db->fd);
    if (indexfile_tmp) {
	amwait_t index_status;

	/*@i@*/ aclose(indexout);
	waitpid(indexpid,&index_status,0);
	if (rename(indexfile_tmp, indexfile_real) != 0) {
	    log_add(L_WARNING, _("could not rename \"%s\" to \"%s\": %s"),
		    indexfile_tmp, indexfile_real, strerror(errno));
	}
	amfree(indexfile_tmp);
	amfree(indexfile_real);
    }

    if(db->compresspid != -1) {
	waitpid(db->compresspid,NULL,0);
    }
    if(db->encryptpid != -1) {
	waitpid(db->encryptpid,NULL,0);
    }

    amfree(errstr);

    return 1;

failed:
    q = squotef("[%s]", errstr);
    putresult(FAILED, "%s %s\n", handle, q);
    amfree(q);

    aclose(db->fd);
    /* kill all child process */
    if (db->compresspid != -1) {
	g_fprintf(stderr,_("%s: kill compress command\n"),get_pname());
	if (kill(db->compresspid, SIGTERM) < 0) {
	    if (errno != ESRCH)
		g_fprintf(stderr,_("%s: can't kill compress command: %s\n"), 
		    get_pname(), strerror(errno));
	}
	else {
	    waitpid(db->compresspid,NULL,0);
	}
    }

    if (db->encryptpid != -1) {
	g_fprintf(stderr,_("%s: kill encrypt command\n"),get_pname());
	if (kill(db->encryptpid, SIGTERM) < 0) {
	    if (errno != ESRCH)
		g_fprintf(stderr,_("%s: can't kill encrypt command: %s\n"), 
		    get_pname(), strerror(errno));
	}
	else {
	    waitpid(db->encryptpid,NULL,0);
	}
    }

    if (indexpid != -1) {
	g_fprintf(stderr,_("%s: kill index command\n"),get_pname());
	if (kill(indexpid, SIGTERM) < 0) {
	    if (errno != ESRCH)
		g_fprintf(stderr,_("%s: can't kill index command: %s\n"), 
		    get_pname(),strerror(errno));
	}
	else {
	    waitpid(indexpid,NULL,0);
	}
    }

    log_start_multiline();
    log_add(L_FAIL, _("%s %s %s %d [%s]"), hostname, qdiskname, dumper_timestamp,
	    level, errstr);
    if (errf) {
	log_msgout(L_FAIL);
    }
    log_end_multiline();

    if (errf) afclose(errf);

    if (indexfile_tmp) {
	unlink(indexfile_tmp);
	amfree(indexfile_tmp);
	amfree(indexfile_real);
    }

    return 0;
}

/*
 * Callback for reads on the mesgfd stream
 */
static void
read_mesgfd(
    void *	cookie,
    void *	buf,
    ssize_t	size)
{
    struct databuf *db = cookie;

    assert(db != NULL);

    switch (size) {
    case -1:
	errstr = newvstrallocf(errstr, _("mesg read: %s"),
	    security_stream_geterror(streams[MESGFD].fd));
	dump_result = 2;
	stop_dump();
	return;

    case 0:
	/*
	 * EOF.  Just shut down the mesg stream.
	 */
	process_dumpeof();
	security_stream_close(streams[MESGFD].fd);
	streams[MESGFD].fd = NULL;
	/*
	 * If the data fd and index fd has also shut down, then we're done.
	 */
	if ((set_datafd == 0 || streams[DATAFD].fd == NULL) && 
	    streams[INDEXFD].fd == NULL)
	    stop_dump();
	return;

    default:
	assert(buf != NULL);
	add_msg_data(buf, (size_t)size);
	security_stream_read(streams[MESGFD].fd, read_mesgfd, cookie);
	break;
    }

    if (ISSET(status, GOT_INFO_ENDLINE) && !ISSET(status, HEADER_DONE)) {
	SET(status, HEADER_DONE);
	/* time to do the header */
	finish_tapeheader(&file);
	if (write_tapeheader(db->fd, &file)) {
	    errstr = newvstrallocf(errstr, _("write_tapeheader: %s"), 
				  strerror(errno));
	    dump_result = 2;
	    stop_dump();
	    return;
	}
	dumpsize += (off_t)DISK_BLOCK_KB;
	headersize += (off_t)DISK_BLOCK_KB;

	if (srvencrypt == ENCRYPT_SERV_CUST) {
	    if (runencrypt(db->fd, &db->encryptpid, srvencrypt) < 0) {
		dump_result = 2;
		stop_dump();
		return;
	    }
	}
	/*
	 * Now, setup the compress for the data output, and start
	 * reading the datafd.
	 */
	if ((srvcompress != COMP_NONE) && (srvcompress != COMP_CUST)) {
	    if (runcompress(db->fd, &db->compresspid, srvcompress) < 0) {
		dump_result = 2;
		stop_dump();
		return;
	    }
	}
	security_stream_read(streams[DATAFD].fd, read_datafd, db);
	set_datafd = 1;
    }

    /*
     * Reset the timeout for future reads
     */
    timeout(conf_dtimeout);
}

/*
 * Callback for reads on the datafd stream
 */
static void
read_datafd(
    void *	cookie,
    void *	buf,
    ssize_t	size)
{
    struct databuf *db = cookie;

    assert(db != NULL);

    /*
     * The read failed.  Error out
     */
    if (size < 0) {
	errstr = newvstrallocf(errstr, _("data read: %s"),
	    security_stream_geterror(streams[DATAFD].fd));
	dump_result = 2;
	stop_dump();
	return;
    }

    /* The header had better be written at this point */
    assert(ISSET(status, HEADER_DONE));

    /*
     * EOF.  Stop and return.
     */
    if (size == 0) {
	databuf_flush(db);
	if (dumpbytes != (off_t)0) {
	    dumpsize += (off_t)1;
	}
	security_stream_close(streams[DATAFD].fd);
	streams[DATAFD].fd = NULL;
	/*
	 * If the mesg fd and index fd has also shut down, then we're done.
	 */
	if (streams[MESGFD].fd == NULL && streams[INDEXFD].fd == NULL)
	    stop_dump();
	return;
    }

    /*
     * We read something.  Add it to the databuf and reschedule for
     * more data.
     */
    assert(buf != NULL);
    if (databuf_write(db, buf, (size_t)size) < 0) {
	errstr = newvstrallocf(errstr, _("data write: %s"), strerror(errno));
	dump_result = 2;
	stop_dump();
	return;
    }

    /*
     * Reset the timeout for future reads
     */
    timeout(conf_dtimeout);

    security_stream_read(streams[DATAFD].fd, read_datafd, cookie);
}

/*
 * Callback for reads on the index stream
 */
static void
read_indexfd(
    void *	cookie,
    void *	buf,
    ssize_t	size)
{
    int fd;

    assert(cookie != NULL);
    fd = *(int *)cookie;

    if (size < 0) {
	errstr = newvstrallocf(errstr, _("index read: %s"),
	    security_stream_geterror(streams[INDEXFD].fd));
	dump_result = 2;
	stop_dump();
	return;
    }

    /*
     * EOF.  Stop and return.
     */
    if (size == 0) {
	security_stream_close(streams[INDEXFD].fd);
	streams[INDEXFD].fd = NULL;
	/*
	 * If the mesg fd has also shut down, then we're done.
	 */
	if ((set_datafd == 0 || streams[DATAFD].fd == NULL) &&
	     streams[MESGFD].fd == NULL)
	    stop_dump();
	return;
    }

    assert(buf != NULL);

    /*
     * We ignore error while writing to the index file.
     */
    if (fullwrite(fd, buf, (size_t)size) < 0) {
	/* Ignore error, but schedule another read. */
	if(indexfderror == 0) {
	    indexfderror = 1;
	    log_add(L_INFO, _("Index corrupted for %s:%s"), hostname, qdiskname);
	}
    }
    security_stream_read(streams[INDEXFD].fd, read_indexfd, cookie);
}

/*
 * Startup a timeout in the event handler.  If the arg is 0,
 * then remove the timeout.
 */
static void
timeout(
    time_t seconds)
{
    static event_handle_t *ev_timeout = NULL;

    /*
     * First, remove a timeout if one is active.
     */
    if (ev_timeout != NULL) {
	event_release(ev_timeout);
	ev_timeout = NULL;
    }

    /*
     * Now, schedule a new one if 'seconds' is greater than 0
     */
    if (seconds > 0)
	ev_timeout = event_register((event_id_t)seconds, EV_TIME, timeout_callback, NULL);
}

/*
 * This is the callback for timeout().  If this is reached, then we
 * have a data timeout.
 */
static void
timeout_callback(
    void *	unused)