2 * Amanda, The Advanced Maryland Automatic Network Disk Archiver
3 * Copyright (c) 2009-2012 Zmanda, Inc. All Rights Reserved.
5 * This program is free software; you can redistribute it and/or modify it
6 * under the terms of the GNU General Public License version 2 as published
7 * by the Free Software Foundation.
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
24 #include "xfer-device.h"
28 /* A transfer destination that writes and entire dumpfile to one or more files
29 * on one or more devices via DirectTCP, handling the work of spanning a
30 * directtcp connection over multiple devices. Note that this assumes the
31 * devices support early EOM warning. */
34 * Xfer Dest Taper DirectTCP
37 static GType xfer_dest_taper_directtcp_get_type(void);
38 #define XFER_DEST_TAPER_DIRECTTCP_TYPE (xfer_dest_taper_directtcp_get_type())
39 #define XFER_DEST_TAPER_DIRECTTCP(obj) G_TYPE_CHECK_INSTANCE_CAST((obj), xfer_dest_taper_directtcp_get_type(), XferDestTaperDirectTCP)
40 #define XFER_DEST_TAPER_DIRECTTCP_CONST(obj) G_TYPE_CHECK_INSTANCE_CAST((obj), xfer_dest_taper_directtcp_get_type(), XferDestTaperDirectTCP const)
41 #define XFER_DEST_TAPER_DIRECTTCP_CLASS(klass) G_TYPE_CHECK_CLASS_CAST((klass), xfer_dest_taper_directtcp_get_type(), XferDestTaperDirectTCPClass)
42 #define IS_XFER_DEST_TAPER_DIRECTTCP(obj) G_TYPE_CHECK_INSTANCE_TYPE((obj), xfer_dest_taper_directtcp_get_type ())
43 #define XFER_DEST_TAPER_DIRECTTCP_GET_CLASS(obj) G_TYPE_INSTANCE_GET_CLASS((obj), xfer_dest_taper_directtcp_get_type(), XferDestTaperDirectTCPClass)
45 static GObjectClass *parent_class = NULL;
47 typedef struct XferDestTaperDirectTCP {
48 XferDestTaper __parent__;
50 /* constructor parameters */
51 guint64 part_size; /* (bytes) */
54 GThread *worker_thread;
56 /* state (governs everything below) */
60 Device *volatile device; /* device to write to (refcounted) */
61 dumpfile_t *volatile part_header;
63 /* did the device listen proceed without error? */
66 /* part number in progress */
67 volatile guint64 partnum;
69 /* connection we're writing to (refcounted) */
70 DirectTCPConnection *conn;
72 /* is the element paused, waiting to start a new part? this is set to FALSE
73 * by the main thread to start a part, and the worker thread waits on the
74 * corresponding condition variable. */
75 volatile gboolean paused;
77 GCond *abort_accept_cond; /* condition to trigger to abort an accept */
79 } XferDestTaperDirectTCP;
82 XferDestTaperClass __parent__;
83 } XferDestTaperDirectTCPClass;
89 #define DBG(LEVEL, ...) if (debug_taper >= LEVEL) { _xdt_dbg(__VA_ARGS__); }
91 _xdt_dbg(const char *fmt, ...)
96 arglist_start(argp, fmt);
97 g_vsnprintf(msg, sizeof(msg), fmt, argp);
99 g_debug("XDT: %s", msg);
109 XferElement *elt = (XferElement *)data;
110 XferDestTaperDirectTCP *self = (XferDestTaperDirectTCP *)data;
111 GTimer *timer = g_timer_new();
114 /* This thread's job is to accept() an incoming connection, then call
115 * write_from_connection for each part, and then close the connection */
117 /* If the device_listen failed, then we will soon be cancelled, so wait
118 * for that to occur and then send XMSG_DONE */
119 if (!self->listen_ok) {
120 DBG(2, "listen failed; waiting for cancellation without attempting an accept");
121 wait_until_xfer_cancelled(elt->xfer);
125 g_mutex_lock(self->state_mutex);
127 /* first, accept a new connection from the device */
128 DBG(2, "accepting DirectTCP connection on device %s", self->device->device_name);
129 result = device_accept_with_cond(self->device, &self->conn,
131 self->abort_accept_cond);
133 xfer_cancel_with_error(XFER_ELEMENT(self),
134 "accepting DirectTCP connection: %s",
135 device_error_or_status(self->device));
136 g_mutex_unlock(self->state_mutex);
138 } else if (result == 1) {
139 g_mutex_unlock(self->state_mutex);
143 DBG(2, "connection accepted; sending XMSG_READY");
144 xfer_queue_message(elt->xfer, xmsg_new(elt, XMSG_READY, 0));
146 /* round the part size up to the next multiple of the block size */
147 if (self->part_size) {
148 self->part_size += self->device->block_size-1;
149 self->part_size -= self->part_size % self->device->block_size;
152 /* now loop until we're out of parts */
159 /* wait to be un-paused */
160 while (!elt->cancelled && self->paused) {
161 DBG(9, "waiting to be un-paused");
162 g_cond_wait(self->paused_cond, self->state_mutex);
164 DBG(9, "done waiting");
169 DBG(2, "writing part to %s", self->device->device_name);
170 if (!device_start_file(self->device, self->part_header) || self->device->is_eom) {
171 /* this is not fatal to the transfer, since no data was lost. We
172 * just need a new device. The scribe special-cases 0-byte parts, and will
173 * not record this in the catalog. */
176 dumpfile_free(self->part_header);
177 self->part_header = NULL;
182 dumpfile_free(self->part_header);
183 self->part_header = NULL;
185 fileno = self->device->file;
186 g_assert(fileno > 0);
189 g_timer_start(timer);
190 if (!device_write_from_connection(self->device,
191 self->part_size, &size)) {
192 /* even if this is just a physical EOM, we may have lost data, so
193 * the whole transfer is dead. */
194 xfer_cancel_with_error(XFER_ELEMENT(self),
195 "Error writing from DirectTCP connection: %s",
196 device_error_or_status(self->device));
201 eom = self->device->is_eom;
202 eof = self->device->is_eof;
204 /* finish the file, even if we're at EOM, but if this fails then we may
206 if (!device_finish_file(self->device)) {
207 xfer_cancel_with_error(XFER_ELEMENT(self),
208 "Error finishing tape file: %s",
209 device_error_or_status(self->device));
213 /* if we wrote zero bytes and reached EOM, then this is an empty part */
214 if (eom && !eof && size == 0) {
218 msg = xmsg_new(XFER_ELEMENT(self), XMSG_PART_DONE, 0);
220 msg->duration = g_timer_elapsed(timer, NULL);
221 msg->partnum = self->partnum;
222 msg->fileno = fileno;
223 msg->successful = TRUE;
227 /* time runs backward on some test boxes, so make sure this is positive */
228 if (msg->duration < 0) msg->duration = 0;
230 xfer_queue_message(elt->xfer, msg);
234 /* we're done at EOF */
238 /* wait to be unpaused again */
243 msg = xmsg_new(XFER_ELEMENT(self), XMSG_PART_DONE, 0);
248 msg->successful = TRUE;
251 xfer_queue_message(elt->xfer, msg);
253 /* wait to be unpaused again */
258 /* drop the mutex and wait until all elements have been cancelled
259 * before closing the connection */
260 g_mutex_unlock(self->state_mutex);
261 wait_until_xfer_cancelled(elt->xfer);
262 g_mutex_lock(self->state_mutex);
266 /* close the DirectTCP connection */
267 directtcp_connection_close(self->conn);
268 g_object_unref(self->conn);
271 g_mutex_unlock(self->state_mutex);
272 g_timer_destroy(timer);
275 xfer_queue_message(elt->xfer, xmsg_new(XFER_ELEMENT(self), XMSG_DONE, 0));
288 XferDestTaperDirectTCP *self = (XferDestTaperDirectTCP *)elt;
290 /* start the device listening, and get the addresses */
291 if (!device_listen(self->device, TRUE, &elt->input_listen_addrs)) {
292 elt->input_listen_addrs = NULL;
293 xfer_cancel_with_error(XFER_ELEMENT(self),
294 "Error starting DirectTCP listen: %s",
295 device_error_or_status(self->device));
296 self->listen_ok = FALSE;
300 self->listen_ok = TRUE;
308 XferDestTaperDirectTCP *self = (XferDestTaperDirectTCP *)elt;
309 GError *error = NULL;
313 /* start up the thread */
314 self->worker_thread = g_thread_create(worker_thread, (gpointer)self, TRUE, &error);
315 if (!self->worker_thread) {
316 g_critical(_("Error creating new thread: %s (%s)"),
317 error->message, errno? strerror(errno) : _("no error code"));
328 XferDestTaperDirectTCP *self = XFER_DEST_TAPER_DIRECTTCP(elt);
332 rv = XFER_ELEMENT_CLASS(parent_class)->cancel(elt, expect_eof);
334 /* signal all of the condition variables to realize that we're no
336 g_mutex_lock(self->state_mutex);
337 g_cond_broadcast(self->paused_cond);
338 g_cond_broadcast(self->abort_accept_cond);
339 g_mutex_unlock(self->state_mutex);
346 XferDestTaper *xdtself,
350 XferDestTaperDirectTCP *self = XFER_DEST_TAPER_DIRECTTCP(xdtself);
352 /* the only way self->device can become NULL is if use_device fails, in
353 * which case an error is already queued up, so just return silently */
354 if (self->device == NULL)
357 g_assert(!self->device->in_file);
358 g_assert(header != NULL);
360 DBG(1, "start_part(retry_part=%d)", retry_part);
362 g_mutex_lock(self->state_mutex);
363 g_assert(self->paused);
365 if (self->part_header)
366 dumpfile_free(self->part_header);
367 self->part_header = dumpfile_copy(header);
370 self->paused = FALSE;
371 g_cond_broadcast(self->paused_cond);
373 g_mutex_unlock(self->state_mutex);
378 XferDestTaper *xdtself,
381 XferDestTaperDirectTCP *self = XFER_DEST_TAPER_DIRECTTCP(xdtself);
383 /* short-circuit if nothing is changing */
384 if (self->device == device)
387 g_mutex_lock(self->state_mutex);
390 g_object_unref(self->device);
393 /* if we already have a connection, then make this device use it */
395 if (!device_use_connection(device, self->conn)) {
396 /* queue up an error for later, and leave the device NULL.
397 * start_part will see this and fail silently */
398 xfer_cancel_with_error(XFER_ELEMENT(self),
399 _("Failed part was not cached; cannot retry"));
404 self->device = device;
405 g_object_ref(device);
407 g_mutex_unlock(self->state_mutex);
411 get_part_bytes_written_impl(
412 XferDestTaper *xdtself G_GNUC_UNUSED)
414 /* This operation is not supported for this taper dest. Maybe someday. */
422 XferDestTaperDirectTCP *self = XFER_DEST_TAPER_DIRECTTCP(elt);
423 elt->can_generate_eof = FALSE;
425 self->worker_thread = NULL;
428 self->state_mutex = g_mutex_new();
429 self->paused_cond = g_cond_new();
430 self->abort_accept_cond = g_cond_new();
437 XferDestTaperDirectTCP *self = XFER_DEST_TAPER_DIRECTTCP(obj_self);
440 g_object_unref(self->conn);
444 g_object_unref(self->device);
448 g_object_unref(self->device);
451 g_mutex_free(self->state_mutex);
452 g_cond_free(self->paused_cond);
453 g_cond_free(self->abort_accept_cond);
455 if (self->part_header)
456 dumpfile_free(self->part_header);
457 self->part_header = NULL;
460 G_OBJECT_CLASS(parent_class)->finalize(obj_self);
465 XferDestTaperDirectTCPClass * selfc)
467 XferElementClass *klass = XFER_ELEMENT_CLASS(selfc);
468 XferDestTaperClass *xdt_klass = XFER_DEST_TAPER_CLASS(selfc);
469 GObjectClass *goc = G_OBJECT_CLASS(selfc);
470 static xfer_element_mech_pair_t mech_pairs[] = {
471 { XFER_MECH_DIRECTTCP_LISTEN, XFER_MECH_NONE, 0, 0},
472 { XFER_MECH_NONE, XFER_MECH_NONE, 0, 0},
475 klass->start = start_impl;
476 klass->setup = setup_impl;
477 klass->cancel = cancel_impl;
478 xdt_klass->start_part = start_part_impl;
479 xdt_klass->use_device = use_device_impl;
480 xdt_klass->get_part_bytes_written = get_part_bytes_written_impl;
481 goc->finalize = finalize_impl;
483 klass->perl_class = "Amanda::Xfer::Dest::Taper::DirectTCP";
484 klass->mech_pairs = mech_pairs;
486 parent_class = g_type_class_peek_parent(selfc);
490 xfer_dest_taper_directtcp_get_type (void)
492 static GType type = 0;
494 if G_UNLIKELY(type == 0) {
495 static const GTypeInfo info = {
496 sizeof (XferDestTaperDirectTCPClass),
497 (GBaseInitFunc) NULL,
498 (GBaseFinalizeFunc) NULL,
499 (GClassInitFunc) class_init,
500 (GClassFinalizeFunc) NULL,
501 NULL /* class_data */,
502 sizeof (XferDestTaperDirectTCP),
504 (GInstanceInitFunc) instance_init,
508 type = g_type_register_static (XFER_DEST_TAPER_TYPE, "XferDestTaperDirectTCP", &info, 0);
519 xfer_dest_taper_directtcp(Device *first_device, guint64 part_size)
521 XferDestTaperDirectTCP *self = (XferDestTaperDirectTCP *)g_object_new(XFER_DEST_TAPER_DIRECTTCP_TYPE, NULL);
523 g_assert(device_directtcp_supported(first_device));
525 self->part_size = part_size;
526 self->device = first_device;
528 g_object_ref(self->device);
530 return XFER_ELEMENT(self);