2 * Amanda, The Advanced Maryland Automatic Network Disk Archiver
3 * Copyright (c) 2009, 2010 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;
78 } XferDestTaperDirectTCP;
81 XferDestTaperClass __parent__;
82 } XferDestTaperDirectTCPClass;
88 #define DBG(LEVEL, ...) if (debug_taper >= LEVEL) { _xdt_dbg(__VA_ARGS__); }
90 _xdt_dbg(const char *fmt, ...)
95 arglist_start(argp, fmt);
96 g_vsnprintf(msg, sizeof(msg), fmt, argp);
98 g_debug("XDT: %s", msg);
108 XferElement *elt = (XferElement *)data;
109 XferDestTaperDirectTCP *self = (XferDestTaperDirectTCP *)data;
110 GTimer *timer = g_timer_new();
112 /* This thread's job is to accept() an incoming connection, then call
113 * write_from_connection for each part, and then close the connection */
115 /* If the device_listen failed, then we will soon be cancelled, so wait
116 * for that to occur and then send XMSG_DONE */
117 if (!self->listen_ok) {
118 DBG(2, "listen failed; waiting for cancellation without attempting an accept");
119 wait_until_xfer_cancelled(elt->xfer);
123 g_mutex_lock(self->state_mutex);
125 /* first, accept a new connection from the device */
126 DBG(2, "accepting DirectTCP connection on device %s", self->device->device_name);
127 if (!device_accept(self->device, &self->conn, NULL, NULL)) {
128 xfer_cancel_with_error(XFER_ELEMENT(self),
129 "accepting DirectTCP connection: %s",
130 device_error_or_status(self->device));
131 g_mutex_unlock(self->state_mutex);
135 DBG(2, "connection accepted; sending XMSG_READY");
136 xfer_queue_message(elt->xfer, xmsg_new(elt, XMSG_READY, 0));
138 /* round the part size up to the next multiple of the block size */
139 if (self->part_size) {
140 self->part_size += self->device->block_size-1;
141 self->part_size -= self->part_size % self->device->block_size;
144 /* now loop until we're out of parts */
151 /* wait to be un-paused */
152 while (!elt->cancelled && self->paused) {
153 DBG(9, "waiting to be un-paused");
154 g_cond_wait(self->paused_cond, self->state_mutex);
156 DBG(9, "done waiting");
161 DBG(2, "writing part to %s", self->device->device_name);
162 if (!device_start_file(self->device, self->part_header) || self->device->is_eom) {
163 /* this is not fatal to the transfer, since no data was lost. We
164 * just need a new device. The scribe special-cases 0-byte parts, and will
165 * not record this in the catalog. */
168 dumpfile_free(self->part_header);
169 self->part_header = NULL;
174 dumpfile_free(self->part_header);
175 self->part_header = NULL;
177 fileno = self->device->file;
178 g_assert(fileno > 0);
181 g_timer_start(timer);
182 if (!device_write_from_connection(self->device,
183 self->part_size, &size)) {
184 /* even if this is just a physical EOM, we may have lost data, so
185 * the whole transfer is dead. */
186 xfer_cancel_with_error(XFER_ELEMENT(self),
187 "Error writing from DirectTCP connection: %s",
188 device_error_or_status(self->device));
193 eom = self->device->is_eom;
194 eof = self->device->is_eof;
196 /* finish the file, even if we're at EOM, but if this fails then we may
198 if (!device_finish_file(self->device)) {
199 xfer_cancel_with_error(XFER_ELEMENT(self),
200 "Error finishing tape file: %s",
201 device_error_or_status(self->device));
205 /* if we wrote zero bytes and reached EOM, then this is an empty part */
206 if (eom && !eof && size == 0) {
210 msg = xmsg_new(XFER_ELEMENT(self), XMSG_PART_DONE, 0);
212 msg->duration = g_timer_elapsed(timer, NULL);
213 msg->partnum = self->partnum;
214 msg->fileno = fileno;
215 msg->successful = TRUE;
219 /* time runs backward on some test boxes, so make sure this is positive */
220 if (msg->duration < 0) msg->duration = 0;
222 xfer_queue_message(elt->xfer, msg);
226 /* we're done at EOF */
230 /* wait to be unpaused again */
235 msg = xmsg_new(XFER_ELEMENT(self), XMSG_PART_DONE, 0);
240 msg->successful = TRUE;
243 xfer_queue_message(elt->xfer, msg);
245 /* wait to be unpaused again */
250 /* drop the mutex and wait until all elements have been cancelled
251 * before closing the connection */
252 g_mutex_unlock(self->state_mutex);
253 wait_until_xfer_cancelled(elt->xfer);
254 g_mutex_lock(self->state_mutex);
258 /* close the DirectTCP connection */
259 directtcp_connection_close(self->conn);
260 g_object_unref(self->conn);
263 g_mutex_unlock(self->state_mutex);
264 g_timer_destroy(timer);
267 xfer_queue_message(elt->xfer, xmsg_new(XFER_ELEMENT(self), XMSG_DONE, 0));
280 XferDestTaperDirectTCP *self = (XferDestTaperDirectTCP *)elt;
282 /* start the device listening, and get the addresses */
283 if (!device_listen(self->device, TRUE, &elt->input_listen_addrs)) {
284 elt->input_listen_addrs = NULL;
285 xfer_cancel_with_error(XFER_ELEMENT(self),
286 "Error starting DirectTCP listen: %s",
287 device_error_or_status(self->device));
288 self->listen_ok = FALSE;
292 self->listen_ok = TRUE;
300 XferDestTaperDirectTCP *self = (XferDestTaperDirectTCP *)elt;
301 GError *error = NULL;
305 /* start up the thread */
306 self->worker_thread = g_thread_create(worker_thread, (gpointer)self, TRUE, &error);
307 if (!self->worker_thread) {
308 g_critical(_("Error creating new thread: %s (%s)"),
309 error->message, errno? strerror(errno) : _("no error code"));
320 XferDestTaperDirectTCP *self = XFER_DEST_TAPER_DIRECTTCP(elt);
324 rv = XFER_ELEMENT_CLASS(parent_class)->cancel(elt, expect_eof);
326 /* signal all of the condition variables to realize that we're no
328 g_mutex_lock(self->state_mutex);
329 g_cond_broadcast(self->paused_cond);
330 g_mutex_unlock(self->state_mutex);
337 XferDestTaper *xdtself,
341 XferDestTaperDirectTCP *self = XFER_DEST_TAPER_DIRECTTCP(xdtself);
343 /* the only way self->device can become NULL is if use_device fails, in
344 * which case an error is already queued up, so just return silently */
345 if (self->device == NULL)
348 g_assert(!self->device->in_file);
349 g_assert(header != NULL);
351 DBG(1, "start_part(retry_part=%d)", retry_part);
353 g_mutex_lock(self->state_mutex);
354 g_assert(self->paused);
356 if (self->part_header)
357 dumpfile_free(self->part_header);
358 self->part_header = dumpfile_copy(header);
361 self->paused = FALSE;
362 g_cond_broadcast(self->paused_cond);
364 g_mutex_unlock(self->state_mutex);
369 XferDestTaper *xdtself,
372 XferDestTaperDirectTCP *self = XFER_DEST_TAPER_DIRECTTCP(xdtself);
374 /* short-circuit if nothing is changing */
375 if (self->device == device)
378 g_mutex_lock(self->state_mutex);
381 g_object_unref(self->device);
384 /* if we already have a connection, then make this device use it */
386 if (!device_use_connection(device, self->conn)) {
387 /* queue up an error for later, and leave the device NULL.
388 * start_part will see this and fail silently */
389 xfer_cancel_with_error(XFER_ELEMENT(self),
390 _("Failed part was not cached; cannot retry"));
395 self->device = device;
396 g_object_ref(device);
398 g_mutex_unlock(self->state_mutex);
402 get_part_bytes_written_impl(
403 XferDestTaper *xdtself G_GNUC_UNUSED)
405 /* This operation is not supported for this taper dest. Maybe someday. */
413 XferDestTaperDirectTCP *self = XFER_DEST_TAPER_DIRECTTCP(elt);
414 elt->can_generate_eof = FALSE;
416 self->worker_thread = NULL;
419 self->state_mutex = g_mutex_new();
420 self->paused_cond = g_cond_new();
427 XferDestTaperDirectTCP *self = XFER_DEST_TAPER_DIRECTTCP(obj_self);
430 g_object_unref(self->conn);
434 g_object_unref(self->device);
438 g_object_unref(self->device);
441 g_mutex_free(self->state_mutex);
442 g_cond_free(self->paused_cond);
444 if (self->part_header)
445 dumpfile_free(self->part_header);
446 self->part_header = NULL;
449 G_OBJECT_CLASS(parent_class)->finalize(obj_self);
454 XferDestTaperDirectTCPClass * selfc)
456 XferElementClass *klass = XFER_ELEMENT_CLASS(selfc);
457 XferDestTaperClass *xdt_klass = XFER_DEST_TAPER_CLASS(selfc);
458 GObjectClass *goc = G_OBJECT_CLASS(selfc);
459 static xfer_element_mech_pair_t mech_pairs[] = {
460 { XFER_MECH_DIRECTTCP_LISTEN, XFER_MECH_NONE, 0, 0},
461 { XFER_MECH_NONE, XFER_MECH_NONE, 0, 0},
464 klass->start = start_impl;
465 klass->setup = setup_impl;
466 klass->cancel = cancel_impl;
467 xdt_klass->start_part = start_part_impl;
468 xdt_klass->use_device = use_device_impl;
469 xdt_klass->get_part_bytes_written = get_part_bytes_written_impl;
470 goc->finalize = finalize_impl;
472 klass->perl_class = "Amanda::Xfer::Dest::Taper::DirectTCP";
473 klass->mech_pairs = mech_pairs;
475 parent_class = g_type_class_peek_parent(selfc);
479 xfer_dest_taper_directtcp_get_type (void)
481 static GType type = 0;
483 if G_UNLIKELY(type == 0) {
484 static const GTypeInfo info = {
485 sizeof (XferDestTaperDirectTCPClass),
486 (GBaseInitFunc) NULL,
487 (GBaseFinalizeFunc) NULL,
488 (GClassInitFunc) class_init,
489 (GClassFinalizeFunc) NULL,
490 NULL /* class_data */,
491 sizeof (XferDestTaperDirectTCP),
493 (GInstanceInitFunc) instance_init,
497 type = g_type_register_static (XFER_DEST_TAPER_TYPE, "XferDestTaperDirectTCP", &info, 0);
508 xfer_dest_taper_directtcp(Device *first_device, guint64 part_size)
510 XferDestTaperDirectTCP *self = (XferDestTaperDirectTCP *)g_object_new(XFER_DEST_TAPER_DIRECTTCP_TYPE, NULL);
512 g_assert(device_directtcp_supported(first_device));
514 self->part_size = part_size;
515 self->device = first_device;
517 g_object_ref(self->device);
519 return XFER_ELEMENT(self);