protocol.c

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

/*
 * Amanda, The Advanced Maryland Automatic Network Disk Archiver
 * Copyright (c) 1991-1999 University of Maryland at College Park
 * All Rights Reserved.
 *
 * Permission to use, copy, modify, distribute, and sell this software and its
 * documentation for any purpose is hereby granted without fee, provided that
 * the above copyright notice appear in all copies and that both that
 * copyright notice and this permission notice appear in supporting
 * documentation, and that the name of U.M. not be used in advertising or
 * publicity pertaining to distribution of the software without specific,
 * written prior permission.  U.M. makes no representations about the
 * suitability of this software for any purpose.  It is provided "as is"
 * without express or implied warranty.
 *
 * U.M. DISCLAIMS ALL WARRANTIES WITH REGARD TO THIS SOFTWARE, INCLUDING ALL
 * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS, IN NO EVENT SHALL U.M.
 * BE LIABLE FOR ANY SPECIAL, INDIRECT OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
 * WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN ACTION
 * OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF OR IN
 * CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
 *
 * Authors: the Amanda Development Team.  Its members are listed in a
 * file named AUTHORS, in the root directory of this distribution.
 */
/*
 * $Id: protocol.c,v 1.45 2006/05/25 17:07:31 martinea Exp $
 *
 * implements amanda protocol
 */
#include "amanda.h"
#include "conffile.h"
#include "event.h"
#include "packet.h"
#include "security.h"
#include "protocol.h"

#define proto_debug(i, ...) do {	\
       if ((i) <= debug_protocol) {	\
           dbprintf(__VA_ARGS__);	\
       }				\
} while (0)

/*
 * Valid actions that can be passed to the state machine
 */
typedef enum {
	PA_START,
	PA_TIMEOUT,
	PA_ERROR,
	PA_RCVDATA,
	PA_CONTPEND,
	PA_PENDING,
	PA_CONTINUE,
	PA_FINISH,
	PA_ABORT
} p_action_t;

/*
 * The current state type.  States are represented as function
 * vectors.
 */
struct proto;
typedef p_action_t (*pstate_t)(struct proto *, p_action_t, pkt_t *);

/*
 * This is a request structure that is wrapped around a packet while it
 * is being passed through amanda.  It holds the timeouts, state, and handles
 * for each request.
 */
typedef struct proto {
    pstate_t state;			/* current state of the request */
    char *hostname;			/* remote host */
    const security_driver_t *security_driver;	/* for connect retries */
    security_handle_t *security_handle;	/* network stream for this req */
    time_t timeout;			/* seconds for this timeout */
    time_t repwait;			/* seconds to wait for reply */
    time_t origtime;			/* orig start time of this request */
    time_t curtime;			/* time when this attempt started */
    int connecttries;			/* times we'll retry a connect */
    int resettries;			/* times we'll resend a REQ */
    int reqtries;			/* times we'll wait for an a ACK */
    pkt_t req;				/* the actual wire request */
    protocol_sendreq_callback continuation; /* call when req dies/finishes */
    void *datap;			/* opaque cookie passed to above */
    char *(*conf_fn)(char *, void *);	/* configuration function */
} proto_t;

#define	CONNECT_WAIT	5	/* secs between connect attempts */
#define ACK_WAIT	10	/* time (secs) to wait for ACK - keep short */
#define RESET_TRIES	2	/* num restarts (reboot/crash) */
#define CURTIME	(time(0) - proto_init_time) /* time relative to start */

/* if no reply in an hour, just forget it */
#define	DROP_DEAD_TIME(t)	(CURTIME - (t) > (60 * 60))

/* get the size of an array */
#define	ASIZE(arr)	(int)(sizeof(arr) / sizeof((arr)[0]))

/*
 * Initialization time
 */
static time_t proto_init_time;

/* local functions */

static const char *action2str(p_action_t);
static const char *pstate2str(pstate_t);

static void connect_callback(void *, security_handle_t *, security_status_t);
static void connect_wait_callback(void *);
static void recvpkt_callback(void *, pkt_t *, security_status_t);

static p_action_t s_sendreq(proto_t *, p_action_t, pkt_t *);
static p_action_t s_ackwait(proto_t *, p_action_t, pkt_t *);
static p_action_t s_repwait(proto_t *, p_action_t, pkt_t *);
static void state_machine(proto_t *, p_action_t, pkt_t *);

/*
 * -------------------
 * Interface functions
 */

/*
 * Initialize globals.
 */
void
protocol_init(void)
{

    proto_init_time = time(NULL);
}

/*
 * Generate a request packet, and submit it to the state machine
 * for transmission.
 */
void
protocol_sendreq(
    const char *		hostname,
    const security_driver_t *	security_driver,
    char *			(*conf_fn)(char *, void *),
    const char *		req,
    time_t			repwait,
    protocol_sendreq_callback	continuation,
    void *			datap)
{
    proto_t *p;

    p = alloc(SIZEOF(proto_t));
    p->state = s_sendreq;
    p->hostname = stralloc(hostname);
    p->security_driver = security_driver;
    /* p->security_handle set in connect_callback */
    p->repwait = repwait;
    p->origtime = CURTIME;
    /* p->curtime set in the sendreq state */
    p->connecttries = getconf_int(CNF_CONNECT_TRIES);
    p->resettries = RESET_TRIES;
    p->reqtries = getconf_int(CNF_REQ_TRIES);
    p->conf_fn = conf_fn;
    pkt_init(&p->req, P_REQ, req);

    /*
     * These are here for the caller
     * We call the continuation function after processing is complete.
     * We pass the datap on through untouched.  It is here so the caller
     * has a way to keep state with each request.
     */
    p->continuation = continuation;
    p->datap = datap;

    proto_debug(1, _("protocol: security_connect: host %s -> p %p\n"), 
		    hostname, p);

    security_connect(p->security_driver, p->hostname, conf_fn, connect_callback,
			 p, p->datap);
}

/*
 * This is a callback for security_connect.  After the security layer
 * has initiated a connection to the given host, this will be called
 * with a security_handle_t.
 *
 * On error, the security_status_t arg will reflect errors which can
 * be had via security_geterror on the handle.
 */
static void
connect_callback(
    void *		cookie,
    security_handle_t *	security_handle,
    security_status_t	status)
{
    proto_t *p = cookie;

    assert(p != NULL);
    p->security_handle = security_handle;

    proto_debug(1, _("protocol: connect_callback: p %p\n"), p);

    switch (status) {
    case S_OK:
	state_machine(p, PA_START, NULL);
	break;

    case S_TIMEOUT:
	security_seterror(p->security_handle, _("timeout during connect"));
	/* FALLTHROUGH */

    case S_ERROR:
	/*
	 * For timeouts or errors, retry a few times, waiting CONNECT_WAIT
	 * seconds between each attempt.  If they all fail, just return
	 * an error back to the caller.
	 */
	if (--p->connecttries == 0) {
	    state_machine(p, PA_ABORT, NULL);
	} else {
	    proto_debug(1, _("protocol: connect_callback: p %p: retrying %s\n"),
			    p, p->hostname);
	    security_close(p->security_handle);
	    /* XXX overload p->security handle to hold the event handle */
	    p->security_handle =
		(security_handle_t *)event_register(CONNECT_WAIT, EV_TIME,
		connect_wait_callback, p);
	}
	break;

    default:
	assert(0);
	break;
    }
}

/*
 * This gets called when a host has been put on a wait queue because
 * initial connection attempts failed.
 */
static void
connect_wait_callback(
    void *	cookie)
{
    proto_t *p = cookie;

    event_release((event_handle_t *)p->security_handle);
    security_connect(p->security_driver, p->hostname, p->conf_fn,
	connect_callback, p, p->datap);
}


/*
 * Does a one pass protocol sweep.  Handles any incoming packets that 
 * are waiting to be processed, and then deals with any pending
 * requests that have timed out.
 *
 * Callers should periodically call this after they have submitted
 * requests if they plan on doing a lot of work.
 */
void
protocol_check(void)
{

    /* arg == 1 means don't block */
    event_loop(1);
}


/*
 * Does an infinite pass protocol sweep.  This doesn't return until all
 * requests have been satisfied or have timed out.
 *
 * Callers should call this after they have finished submitting requests
 * and are just waiting for all of the answers to come back.
 */
void
protocol_run(void)
{

    /* arg == 0 means block forever until no more events are left */
    event_loop(0);
}


/*
 * ------------------
 * Internal functions
 */

/*
 * The guts of the protocol.  This handles the many paths a request can
 * make, including retrying the request and acknowledgements, and dealing
 * with timeouts and successfull replies.
 */
static void
state_machine(
    proto_t *	p,
    p_action_t	action,
    pkt_t *	pkt)
{
    pstate_t curstate;
    p_action_t retaction;

    proto_debug(1, _("protocol: state_machine: initial: p %p action %s pkt %p\n"),
		    p, action2str(action), (void *)NULL);

    assert(p != NULL);
    assert(action == PA_RCVDATA || pkt == NULL);
    assert(p->state != NULL);

    for (;;) {
	proto_debug(1, _("protocol: state_machine: p %p state %s action %s\n"),
			p, pstate2str(p->state), action2str(action));
	if (pkt != NULL) {
	    proto_debug(1, _("protocol: pkt: %s (t %d) orig REQ (t %d cur %d)\n"),
			    pkt_type2str(pkt->type), (int)CURTIME,
			    (int)p->origtime, (int)p->curtime);
	    proto_debug(1, _("protocol: pkt contents:\n-----\n%s-----\n"),
			    pkt->body);
	}

	/*
	 * p->state is a function pointer to the current state a request
	 * is in.
	 *
	 * We keep track of the last state we were in so we can make
	 * sure states which return PA_CONTINUE really have transitioned
	 * the request to a new state.
	 */
	curstate = p->state;

	if (action == PA_ABORT)
	    /*
	     * If the passed action indicates a terminal error, then we
	     * need to move to abort right away.
	     */
	    retaction = PA_ABORT;
	else
	    /*
	     * Else we run the state and perform the action it
	     * requests.
	     */
	    retaction = (*curstate)(p, action, pkt);

	proto_debug(1, _("protocol: state_machine: p %p state %s returned %s\n"),
			p, pstate2str(p->state), action2str(retaction));

	/*
	 * The state function is expected to return one of the following
	 * p_action_t's.
	 */
	switch (retaction) {