git.gag.com Git - debian/amanda/blob - device-src/xfer-dest-taper-directtcp.c

   1 /*
   2  * Amanda, The Advanced Maryland Automatic Network Disk Archiver
   3  * Copyright (c) 2009, 2010 Zmanda, Inc.  All Rights Reserved.
   4  *
   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.
   8  *
   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
  12  * for more details.
  13  *
  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
  17  *
  18  * Contact information: Zmanda Inc., 465 S. Mathilda Ave., Suite 300
  19  * Sunnyvale, CA 94085, USA, or: http://www.zmanda.com
  20  */
  21
  22 #include "amanda.h"
  23 #include "amxfer.h"
  24 #include "xfer-device.h"
  25 #include "arglist.h"
  26 #include "conffile.h"
  27
  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. */
  32
  33 /*
  34  * Xfer Dest Taper DirectTCP
  35  */
  36
  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)
  44
  45 static GObjectClass *parent_class = NULL;
  46
  47 typedef struct XferDestTaperDirectTCP {
  48     XferDestTaper __parent__;
  49
  50     /* constructor parameters */
  51     guint64 part_size; /* (bytes) */
  52
  53     /* thread */
  54     GThread *worker_thread;
  55
  56     /* state (governs everything below) */
  57     GMutex *state_mutex;
  58
  59     /* part parameters */
  60     Device *volatile device; /* device to write to (refcounted) */
  61     dumpfile_t *volatile part_header;
  62
  63     /* did the device listen proceed without error? */
  64     gboolean listen_ok;
  65
  66     /* part number in progress */
  67     volatile guint64 partnum;
  68
  69     /* connection we're writing to (refcounted) */
  70     DirectTCPConnection *conn;
  71
  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;
  76     GCond *paused_cond;
  77
  78 } XferDestTaperDirectTCP;
  79
  80 typedef struct {
  81     XferDestTaperClass __parent__;
  82 } XferDestTaperDirectTCPClass;
  83
  84 /*
  85  * Debug logging
  86  */
  87
  88 #define DBG(LEVEL, ...) if (debug_taper >= LEVEL) { _xdt_dbg(__VA_ARGS__); }
  89 static void
  90 _xdt_dbg(const char *fmt, ...)
  91 {
  92     va_list argp;
  93     char msg[1024];
  94
  95     arglist_start(argp, fmt);
  96     g_vsnprintf(msg, sizeof(msg), fmt, argp);
  97     arglist_end(argp);
  98     g_debug("XDT: %s", msg);
  99 }
 100 /*
 101  * Worker Thread
 102  */
 103
 104 static gpointer
 105 worker_thread(
 106     gpointer data)
 107 {
 108     XferElement *elt = (XferElement *)data;
 109     XferDestTaperDirectTCP *self = (XferDestTaperDirectTCP *)data;
 110     GTimer *timer = g_timer_new();
 111
 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 */
 114
 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);
 120         goto send_xmsg_done;
 121     }
 122
 123     g_mutex_lock(self->state_mutex);
 124
 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);
 132         return NULL;
 133     }
 134
 135     DBG(2, "connection accepted; sending XMSG_READY");
 136     xfer_queue_message(elt->xfer, xmsg_new(elt, XMSG_READY, 0));
 137
 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;
 142     }
 143
 144     /* now loop until we're out of parts */
 145     while (1) {
 146         guint64 size;
 147         int fileno;
 148         XMsg *msg = NULL;
 149         gboolean eom, eof;
 150
 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);
 155         }
 156         DBG(9, "done waiting");
 157
 158         if (elt->cancelled)
 159             break;
 160
 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. */
 166
 167             /* clean up */
 168             dumpfile_free(self->part_header);
 169             self->part_header = NULL;
 170
 171             goto empty_part;
 172         }
 173
 174         dumpfile_free(self->part_header);
 175         self->part_header = NULL;
 176
 177         fileno = self->device->file;
 178         g_assert(fileno > 0);
 179
 180         /* write the part */
 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));
 189             goto cancelled;
 190         }
 191         g_timer_stop(timer);
 192
 193         eom = self->device->is_eom;
 194         eof = self->device->is_eof;
 195
 196         /* finish the file, even if we're at EOM, but if this fails then we may
 197          * have lost data */
 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));
 202             goto cancelled;
 203         }
 204
 205         /* if we wrote zero bytes and reached EOM, then this is an empty part */
 206         if (eom && !eof && size == 0) {
 207             goto empty_part;
 208         }
 209
 210         msg = xmsg_new(XFER_ELEMENT(self), XMSG_PART_DONE, 0);
 211         msg->size = size;
 212         msg->duration = g_timer_elapsed(timer, NULL);
 213         msg->partnum = self->partnum;
 214         msg->fileno = fileno;
 215         msg->successful = TRUE;
 216         msg->eom = eom;
 217         msg->eof = eof;
 218
 219         /* time runs backward on some test boxes, so make sure this is positive */
 220         if (msg->duration < 0) msg->duration = 0;
 221
 222         xfer_queue_message(elt->xfer, msg);
 223
 224         self->partnum++;
 225
 226         /* we're done at EOF */
 227         if (eof)
 228             break;
 229
 230         /* wait to be unpaused again */
 231         self->paused = TRUE;
 232         continue;
 233
 234 empty_part:
 235         msg = xmsg_new(XFER_ELEMENT(self), XMSG_PART_DONE, 0);
 236         msg->size = 0;
 237         msg->duration = 0;
 238         msg->partnum = 0;
 239         msg->fileno = 0;
 240         msg->successful = TRUE;
 241         msg->eom = TRUE;
 242         msg->eof = FALSE;
 243         xfer_queue_message(elt->xfer, msg);
 244
 245         /* wait to be unpaused again */
 246         self->paused = TRUE;
 247         continue;
 248
 249 cancelled:
 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);
 255         break;
 256     }
 257
 258     /* close the DirectTCP connection */
 259     directtcp_connection_close(self->conn);
 260     g_object_unref(self->conn);
 261     self->conn = NULL;
 262
 263     g_mutex_unlock(self->state_mutex);
 264     g_timer_destroy(timer);
 265
 266 send_xmsg_done:
 267     xfer_queue_message(elt->xfer, xmsg_new(XFER_ELEMENT(self), XMSG_DONE, 0));
 268
 269     return NULL;
 270 }
 271
 272 /*
 273  * Element mechanics
 274  */
 275
 276 static gboolean
 277 setup_impl(
 278     XferElement *elt)
 279 {
 280     XferDestTaperDirectTCP *self = (XferDestTaperDirectTCP *)elt;
 281
 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;
 289         return FALSE;
 290     }
 291
 292     self->listen_ok = TRUE;
 293     return TRUE;
 294 }
 295
 296 static gboolean
 297 start_impl(
 298     XferElement *elt)
 299 {
 300     XferDestTaperDirectTCP *self = (XferDestTaperDirectTCP *)elt;
 301     GError *error = NULL;
 302
 303     self->paused = TRUE;
 304
 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"));
 310     }
 311
 312     return TRUE;
 313 }
 314
 315 static gboolean
 316 cancel_impl(
 317     XferElement *elt,
 318     gboolean expect_eof)
 319 {
 320     XferDestTaperDirectTCP *self = XFER_DEST_TAPER_DIRECTTCP(elt);
 321     gboolean rv;
 322
 323     /* chain up first */
 324     rv = XFER_ELEMENT_CLASS(parent_class)->cancel(elt, expect_eof);
 325
 326     /* signal all of the condition variables to realize that we're no
 327      * longer paused */
 328     g_mutex_lock(self->state_mutex);
 329     g_cond_broadcast(self->paused_cond);
 330     g_mutex_unlock(self->state_mutex);
 331
 332     return rv;
 333 }
 334
 335 static void
 336 start_part_impl(
 337     XferDestTaper *xdtself,
 338     gboolean retry_part,
 339     dumpfile_t *header)
 340 {
 341     XferDestTaperDirectTCP *self = XFER_DEST_TAPER_DIRECTTCP(xdtself);
 342
 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)
 346         return;
 347
 348     g_assert(!self->device->in_file);
 349     g_assert(header != NULL);
 350
 351     DBG(1, "start_part(retry_part=%d)", retry_part);
 352
 353     g_mutex_lock(self->state_mutex);
 354     g_assert(self->paused);
 355
 356     if (self->part_header)
 357         dumpfile_free(self->part_header);
 358     self->part_header = dumpfile_copy(header);
 359
 360     DBG(1, "unpausing");
 361     self->paused = FALSE;
 362     g_cond_broadcast(self->paused_cond);
 363
 364     g_mutex_unlock(self->state_mutex);
 365 }
 366
 367 static void
 368 use_device_impl(
 369     XferDestTaper *xdtself,
 370     Device *device)
 371 {
 372     XferDestTaperDirectTCP *self = XFER_DEST_TAPER_DIRECTTCP(xdtself);
 373
 374     /* short-circuit if nothing is changing */
 375     if (self->device == device)
 376         return;
 377
 378     g_mutex_lock(self->state_mutex);
 379
 380     if (self->device)
 381         g_object_unref(self->device);
 382     self->device = NULL;
 383
 384     /* if we already have a connection, then make this device use it */
 385     if (self->conn) {
 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"));
 391             return;
 392         }
 393     }
 394
 395     self->device = device;
 396     g_object_ref(device);
 397
 398     g_mutex_unlock(self->state_mutex);
 399 }
 400
 401 static guint64
 402 get_part_bytes_written_impl(
 403     XferDestTaper *xdtself G_GNUC_UNUSED)
 404 {
 405     /* This operation is not supported for this taper dest.  Maybe someday. */
 406     return 0;
 407 }
 408
 409 static void
 410 instance_init(
 411     XferElement *elt)
 412 {
 413     XferDestTaperDirectTCP *self = XFER_DEST_TAPER_DIRECTTCP(elt);
 414     elt->can_generate_eof = FALSE;
 415
 416     self->worker_thread = NULL;
 417     self->paused = TRUE;
 418     self->conn = NULL;
 419     self->state_mutex = g_mutex_new();
 420     self->paused_cond = g_cond_new();
 421 }
 422
 423 static void
 424 finalize_impl(
 425     GObject * obj_self)
 426 {
 427     XferDestTaperDirectTCP *self = XFER_DEST_TAPER_DIRECTTCP(obj_self);
 428
 429     if (self->conn)
 430         g_object_unref(self->conn);
 431     self->conn = NULL;
 432
 433     if (self->device)
 434         g_object_unref(self->device);
 435     self->device = NULL;
 436
 437     if (self->device)
 438         g_object_unref(self->device);
 439     self->device = NULL;
 440
 441     g_mutex_free(self->state_mutex);
 442     g_cond_free(self->paused_cond);
 443
 444     if (self->part_header)
 445         dumpfile_free(self->part_header);
 446     self->part_header = NULL;
 447
 448     /* chain up */
 449     G_OBJECT_CLASS(parent_class)->finalize(obj_self);
 450 }
 451
 452 static void
 453 class_init(
 454     XferDestTaperDirectTCPClass * selfc)
 455 {
 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},
 462     };
 463
 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;
 471
 472     klass->perl_class = "Amanda::Xfer::Dest::Taper::DirectTCP";
 473     klass->mech_pairs = mech_pairs;
 474
 475     parent_class = g_type_class_peek_parent(selfc);
 476 }
 477
 478 static GType
 479 xfer_dest_taper_directtcp_get_type (void)
 480 {
 481     static GType type = 0;
 482
 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),
 492             0 /* n_preallocs */,
 493             (GInstanceInitFunc) instance_init,
 494             NULL
 495         };
 496
 497         type = g_type_register_static (XFER_DEST_TAPER_TYPE, "XferDestTaperDirectTCP", &info, 0);
 498     }
 499
 500     return type;
 501 }
 502
 503 /*
 504  * Constructor
 505  */
 506
 507 XferElement *
 508 xfer_dest_taper_directtcp(Device *first_device, guint64 part_size)
 509 {
 510     XferDestTaperDirectTCP *self = (XferDestTaperDirectTCP *)g_object_new(XFER_DEST_TAPER_DIRECTTCP_TYPE, NULL);
 511
 512     g_assert(device_directtcp_supported(first_device));
 513
 514     self->part_size = part_size;
 515     self->device = first_device;
 516     self->partnum = 1;
 517     g_object_ref(self->device);
 518
 519     return XFER_ELEMENT(self);
 520 }