2 * Copyright (c) 2008-2012 Zmanda, Inc. All Rights Reserved.
4 * This program is free software; you can redistribute it and/or
5 * modify it under the terms of the GNU General Public License
6 * as published by the Free Software Foundation; either version 2
7 * of the License, or (at your option) any later version.
9 * This program is distributed in the hope that it will be useful, but
10 * WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY
11 * or FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License
14 * You should have received a copy of the GNU General Public License along
15 * with this program; if not, write to the Free Software Foundation, Inc.,
16 * 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
18 * Contact information: Zmanda Inc., 465 S. Mathilda Ave., Suite 300
19 * Sunnyvale, CA 94085, USA, or: http://www.zmanda.com
22 /* An Xfer abstracts an active data transfer through the Amanda core.
30 /* An Xfer represents a flow of data from a source, via zero or more filters,
31 * to a destination. Sources, filters, and destinations are "transfer elements".
32 * The job of the Xfer is glue together a sequence of elements, and provide a
33 * dispatch point for messages from those elements to the caller.
35 * Xfers are not implemented as GObjects because there is no reason to subclass an
36 * Xfer or apply any of the other features that come with GObject.
39 /* The moment-to-moment state of a transfer */
41 XFER_INIT = 1, /* initializing */
42 XFER_START = 2, /* starting */
43 XFER_RUNNING = 3, /* data flowing */
44 XFER_CANCELLING = 4,/* cancellation begun */
45 XFER_CANCELLED = 5, /* all elements cancelled; draining data */
46 XFER_DONE = 6, /* data no longer flowing */
49 /* forward declarations */
59 /* The current status of this transfer. This is read-only, and
60 * must only be accessed from the main thread or with status_mutex
64 /* lock this while checking status in a thread
65 * other than the main thread */
68 /* and wait on this for status changes */
71 /* -- remaining fields are private -- */
75 /* All transfer elements for this transfer, in order from
76 * source to destination. This is initialized when the Xfer is
80 /* temporary string for a representation of this transfer */
83 /* GSource and queue for incoming messages */
84 struct XMsgSource *msg_source;
87 /* Number of active elements remaining (a.k.a. the number of
88 * XMSG_DONE messages to expect) */
89 gint num_active_elements;
91 /* Used to coordinate handing off file descriptors among elements of this
98 typedef struct Xfer Xfer;
100 /* Note that all functions must be called from the main thread unless
103 /* Create a new Xfer object, which should later be freed with xfref_free.
105 * This function adds a reference to each element. The caller should
106 * unreference the elements if it does not intend to use them directly.
107 * The Xfer returned has a refcount of one.
109 * @param elements: array of pointers to transfer elements, in order from source
111 * @param nelements: length of 'elements'
112 * @returns: new Xfer object
114 Xfer *xfer_new(struct XferElement **elements, unsigned int nelements);
116 /* Increase the reference count of a transfer.
118 * @param xfer: the transfer
120 void xfer_ref(Xfer *xfer);
122 /* Decrease the reference count of a transfer, possibly freeing it. A running
123 * transfer (state neither XFER_INIT nor XFER_DONE) will not be freed.
125 * @param xfer: the transfer
127 void xfer_unref(Xfer *xfer);
129 /* Get a GSource which will produce events corresponding to messages from
130 * this transfer. This is a "peek" operation, so the reference count for the
131 * GSource is not affected. Note that the same GSource is returned on every
132 * call for a particular transfer.
134 * @returns: GSource object
136 GSource *xfer_get_source(Xfer *xfer);
138 /* Typedef for the callback to be set on the GSource returned from
141 typedef void (*XMsgCallback)(gpointer data, struct XMsg *msg, Xfer *xfer);
143 /* Queue a message for delivery via this transfer's GSource. This can
144 * be called in any thread.
146 * @param xfer: the transfer
147 * @param msg: the message to queue
149 void xfer_queue_message(Xfer *xfer, struct XMsg *msg);
151 /* Get a representation of this transfer. The string belongs to the transfer, and
152 * will be freed when the transfer is freed.
154 * @param xfer: the Xfer object
155 * @returns: statically allocated string
157 char *xfer_repr(Xfer *xfer);
159 /* Start a transfer. This function will fail with an error message if it is
160 * unable to set up the transfer (e.g., if the elements cannot be connected
163 * @param xfer: the Xfer object
164 * @param offset: the offset to start the transfer from (must be 0)
165 * @param size: the Xfer object: the number of bytes to transfer.
167 void xfer_start(Xfer *xfer, gint64 offset, gint64 size);
169 /* Abort a running transfer. This essentially tells the source to stop
170 * producing data and allows the remainder of the transfer to "drain". Thus
171 * the transfer will signal its completion "normally" some time after
172 * xfer_cancel returns. In particular, the state transitions will occur
176 * - xfer_cancel() (note state may still be XFER_RUNNING on return)
178 * - (individual elements' cancel() methods are invoked)
180 * - (data drains from the transfer)
183 * This function can be called from any thread at any time. It will return
186 * @param xfer: the Xfer object
188 void xfer_cancel(Xfer *xfer);
194 /* Wait for the xfer's state to become CANCELLED or DONE; this is useful to
195 * wait until a cancelletion is in progress before returning an EOF or
196 * otherwise handling a failure. If you call this in the main thread, you'll
197 * be waiting for a while.
199 * @param xfer: the transfer object
200 * @returns: the new status (XFER_CANCELLED or XFER_DONE)
202 xfer_status wait_until_xfer_cancelled(Xfer *xfer);
204 /* Wait for the xfer's state to become anything but START; this is
205 * called *automatically* for every xfer_element_pull_buffer call, as the
206 * upstream element may not be running and ready for a pull just yet. But
207 * the function may be useful in other places, too.
209 * @param xfer: the transfer object
210 * @returns: the new status (XFER_CANCELLED or XFER_DONE)
212 xfer_status wait_until_xfer_running(Xfer *xfer);
214 /* Send an XMSG_ERROR constructed with the given format and arguments, then
215 * cancel the transfer, then wait until the transfer is completely cancelled.
216 * This is the most common error-handling process for transfer elements. All
217 * that remains to be done on return is to branch to the appropriate point in
218 * the cancellation-handling portion of the transfer.
220 * @param elt: the transfer element producing the error
221 * @param fmt: the format for the error message
222 * @param ...: arguments corresponding to the format
224 void xfer_cancel_with_error(struct XferElement *elt, const char *fmt, ...)
227 /* Return the fd in *FDP and set *FDP to NEWFD, all in one step. The operation
228 * is atomic with respect to all other such operations in this transfer, making
229 * this a good way to "move" a file descriptor from one element to another. If
230 * xfer is NULL, the operation proceeds with no locking.
232 * @param xfer: the xfer within which this fd is used
233 * @param fdp: pointer to the file descriptor to swap
234 * @param newfd: the new value for *FDP
235 * @returns: the previous contents of *fdp (may be -1)
237 gint xfer_atomic_swap_fd(Xfer *xfer, gint *fdp, gint newfd);