git.gag.com Git - debian/amanda/blob - device-src/ndmp-device.c

   1 /*
   2  * Copyright (c) 2009-2012 Zmanda, Inc.  All Rights Reserved.
   3  *
   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.
   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 "util.h"
  24 #include "device.h"
  25 #include "directtcp.h"
  26 #include "stream.h"
  27 #include "ndmlib.h"
  28 #include "ndmpconnobj.h"
  29 #include "sockaddr-util.h"
  30
  31 /*
  32  * Type checking and casting macros
  33  */
  34 #define TYPE_NDMP_DEVICE        (ndmp_device_get_type())
  35 #define NDMP_DEVICE(obj)        G_TYPE_CHECK_INSTANCE_CAST((obj), ndmp_device_get_type(), NdmpDevice)
  36 #define NDMP_DEVICE_CONST(obj)  G_TYPE_CHECK_INSTANCE_CAST((obj), ndmp_device_get_type(), NdmpDevice const)
  37 #define NDMP_DEVICE_CLASS(klass)        G_TYPE_CHECK_CLASS_CAST((klass), ndmp_device_get_type(), NdmpDeviceClass)
  38 #define IS_NDMP_DEVICE(obj)     G_TYPE_CHECK_INSTANCE_TYPE((obj), ndmp_device_get_type ())
  39 #define NDMP_DEVICE_GET_CLASS(obj)      G_TYPE_INSTANCE_GET_CLASS((obj), ndmp_device_get_type(), NdmpDeviceClass)
  40 static GType    ndmp_device_get_type    (void);
  41
  42 /*
  43  * Main object structure
  44  */
  45
  46 typedef struct NdmpDevice_ NdmpDevice;
  47 struct NdmpDevice_ {
  48     Device __parent__;
  49
  50     NDMPConnection *ndmp;
  51
  52     /* true if tape service is open on the NDMP server */
  53     gboolean tape_open;
  54
  55     /* addresses the object is listening on, and how the connection
  56      * was opened */
  57     DirectTCPAddr *listen_addrs;
  58     gboolean for_writing;
  59
  60     /* support for IndirectTCP */
  61     int indirecttcp_sock; /* -1 if not in use */
  62     int indirect;
  63
  64     /* Current DirectTCPConnectionNDMP */
  65     struct DirectTCPConnectionNDMP_ *directtcp_conn;
  66
  67     /* constructor parameters and properties */
  68     gchar        *ndmp_hostname;
  69     gint         ndmp_port;
  70     gchar        *ndmp_device_name;
  71     gchar        *ndmp_username;
  72     gchar        *ndmp_password;
  73     gchar        *ndmp_auth;
  74     gboolean     verbose;
  75     gsize        read_block_size;
  76
  77     GMutex      *abort_mutex;
  78     GCond       *abort_cond;
  79     gboolean     cancel;
  80     int         *cancelled;
  81 };
  82
  83 /*
  84  * Class definition
  85  */
  86
  87 typedef struct NdmpDeviceClass_ NdmpDeviceClass;
  88 struct NdmpDeviceClass_ {
  89     DeviceClass __parent__;
  90 };
  91
  92 /*
  93  * A directtcp connection subclass representing a running mover on the other end of
  94  * the given NDMP connection
  95  */
  96
  97 #define TYPE_DIRECTTCP_CONNECTION_NDMP  (directtcp_connection_ndmp_get_type())
  98 #define DIRECTTCP_CONNECTION_NDMP(obj)  G_TYPE_CHECK_INSTANCE_CAST((obj), directtcp_connection_ndmp_get_type(), DirectTCPConnectionNDMP)
  99 #define DIRECTTCP_CONNECTION_NDMP_CONST(obj)    G_TYPE_CHECK_INSTANCE_CAST((obj), directtcp_connection_ndmp_get_type(), DirectTCPConnectionNDMP const)
 100 #define DIRECTTCP_CONNECTION_NDMP_CLASS(klass)  G_TYPE_CHECK_CLASS_CAST((klass), directtcp_connection_ndmp_get_type(), DirectTCPConnectionNDMPClass)
 101 #define IS_DIRECTTCP_CONNECTION_NDMP(obj)       G_TYPE_CHECK_INSTANCE_TYPE((obj), directtcp_connection_ndmp_get_type ())
 102 #define DIRECTTCP_CONNECTION_NDMP_GET_CLASS(obj)        G_TYPE_INSTANCE_GET_CLASS((obj), directtcp_connection_ndmp_get_type(), DirectTCPConnectionNDMPClass)
 103 GType directtcp_connection_ndmp_get_type(void);
 104
 105 typedef struct DirectTCPConnectionNDMP_ {
 106     DirectTCPConnection __parent__;
 107
 108     /* NDMP connection controlling the mover */
 109     NDMPConnection *ndmp;
 110
 111     /* mode for this operation */
 112     ndmp9_mover_mode mode;
 113
 114     /* last reported mover position in the datastream */
 115     guint64 offset;
 116 } DirectTCPConnectionNDMP;
 117
 118 typedef struct DirectTCPConnectionNDMPClass_ {
 119     DirectTCPConnectionClass __parent__;
 120 } DirectTCPConnectionNDMPClass;
 121
 122 static DirectTCPConnectionNDMP *directtcp_connection_ndmp_new(
 123         NDMPConnection *ndmp,
 124         ndmp9_mover_mode mode);
 125
 126 /*
 127  * Constants and static data
 128  */
 129
 130 #define NDMP_DEVICE_NAME "ndmp"
 131
 132 /* pointer to the class of our parent */
 133 static DeviceClass *parent_class = NULL;
 134
 135 /* robust_write results */
 136 typedef enum {
 137     ROBUST_WRITE_OK,
 138     ROBUST_WRITE_OK_LEOM,
 139     ROBUST_WRITE_ERROR, /* device error already set */
 140     ROBUST_WRITE_NO_SPACE
 141 } robust_write_result;
 142
 143 /*
 144  * device-specific properties
 145  */
 146
 147 /* Authentication information for NDMP agent. Both of these are strings. */
 148 static DevicePropertyBase device_property_ndmp_username;
 149 static DevicePropertyBase device_property_ndmp_password;
 150 static DevicePropertyBase device_property_ndmp_auth;
 151 static DevicePropertyBase device_property_indirect;
 152 #define PROPERTY_NDMP_USERNAME (device_property_ndmp_username.ID)
 153 #define PROPERTY_NDMP_PASSWORD (device_property_ndmp_password.ID)
 154 #define PROPERTY_NDMP_AUTH (device_property_ndmp_auth.ID)
 155 #define PROPERTY_INDIRECT (device_property_indirect.ID)
 156
 157
 158 /*
 159  * prototypes
 160  */
 161
 162 void ndmp_device_register(void);
 163 static void set_error_from_ndmp(NdmpDevice *self);
 164
 165 #define ndmp_device_read_size(self) \
 166     (((NdmpDevice *)(self))->read_block_size? \
 167         ((NdmpDevice *)(self))->read_block_size : ((Device *)(self))->block_size)
 168
 169 /*
 170  * Utility functions
 171  */
 172
 173 static gboolean
 174 open_connection(
 175         NdmpDevice *self)
 176 {
 177     if (!self->ndmp) {
 178         self->ndmp = ndmp_connection_new(
 179             self->ndmp_hostname,
 180             self->ndmp_port,
 181             self->ndmp_username,
 182             self->ndmp_password,
 183             self->ndmp_auth);
 184
 185         if (ndmp_connection_err_code(self->ndmp)) {
 186             char *errmsg = ndmp_connection_err_msg(self->ndmp);
 187             device_set_error(DEVICE(self),
 188                 g_strdup_printf("could not connect to ndmp-server '%s:%d': %s",
 189                     self->ndmp_hostname, self->ndmp_port, errmsg),
 190                 DEVICE_STATUS_DEVICE_ERROR);
 191             g_object_unref(self->ndmp);
 192             self->ndmp = NULL;
 193             return FALSE;
 194         }
 195
 196         if (self->verbose)
 197             ndmp_connection_set_verbose(self->ndmp, TRUE);
 198
 199         self->tape_open = FALSE;
 200     }
 201
 202     return TRUE;
 203 }
 204
 205 static void
 206 close_connection(
 207         NdmpDevice *self)
 208 {
 209     /* note that this does not send NDMP_TAPE_CLOSE, as it's used in error
 210      * situations too */
 211     if (self->ndmp) {
 212         g_object_unref(self->ndmp);
 213         self->ndmp = NULL;
 214         self->tape_open = FALSE;
 215     }
 216 }
 217
 218 static gboolean
 219 open_tape_agent(
 220     NdmpDevice *self)
 221 {
 222     guint64 file_num, blockno, blocksize;
 223
 224     /* if already open, stop now */
 225     if (self->tape_open) {
 226         return TRUE;
 227     }
 228
 229     if (!open_connection(self)) {
 230         /* error message set by open_connection */
 231         return FALSE;
 232     }
 233
 234     g_debug("opening tape device '%s' on NDMP server '%s:%d'",
 235         self->ndmp_device_name, self->ndmp_hostname, self->ndmp_port);
 236
 237     /* send NDMP_TAPE_OPEN, using RAW mode so that it will open even with no tape */
 238     if (!ndmp_connection_tape_open(self->ndmp,
 239                 self->ndmp_device_name, NDMP9_TAPE_RAW_MODE)) {
 240         set_error_from_ndmp(self);
 241         return FALSE;
 242     }
 243
 244     /* check that the block sizes match */
 245     if (!ndmp_connection_tape_get_state(self->ndmp,
 246         &blocksize, &file_num, &blockno)) {
 247         set_error_from_ndmp(self);
 248         return FALSE;
 249     }
 250     if (blocksize != 0 && blocksize != DEVICE(self)->block_size) {
 251         device_set_error(DEVICE(self),
 252             g_strdup_printf("NDMP device has fixed block size %ju, but Amanda "
 253                     "device is configured with blocksize %ju", (uintmax_t)blocksize,
 254                     (uintmax_t)(DEVICE(self)->block_size)),
 255             DEVICE_STATUS_DEVICE_ERROR);
 256     }
 257
 258     self->tape_open = TRUE;
 259
 260     return TRUE;
 261 }
 262
 263 static gboolean
 264 close_tape_agent(
 265         NdmpDevice *self)
 266 {
 267     if (self->tape_open) {
 268         g_debug("closing tape device '%s' on NDMP server '%s:%d'",
 269             self->ndmp_device_name, self->ndmp_hostname, self->ndmp_port);
 270         self->tape_open = FALSE; /* count it as closed even if there is an error */
 271         if (!ndmp_connection_tape_close(self->ndmp)) {
 272             set_error_from_ndmp(self);
 273             return FALSE;
 274         }
 275     }
 276
 277     return TRUE;
 278 }
 279
 280 static gboolean
 281 single_ndmp_mtio(
 282     NdmpDevice *self,
 283     ndmp9_tape_mtio_op tape_op)
 284 {
 285     guint resid;
 286
 287     if (!ndmp_connection_tape_mtio(self->ndmp, tape_op, 1, &resid)) {
 288         set_error_from_ndmp(self);
 289         return FALSE;
 290     }
 291
 292     if (resid != 0) {
 293         device_set_error(DEVICE(self),
 294                 g_strdup_printf("NDMP MTIO operation %d did not complete", tape_op),
 295                 DEVICE_STATUS_DEVICE_ERROR);
 296     }
 297
 298     return TRUE;
 299 }
 300
 301 /* get the tape state straight from the device; we try to track these things
 302  * accurately in the device, but sometimes it's good to check. */
 303 static gboolean
 304 ndmp_get_state(
 305     NdmpDevice *self)
 306 {
 307     Device *dself = DEVICE(self);
 308     guint64 file_num, blockno, blocksize;
 309
 310     if (!ndmp_connection_tape_get_state(self->ndmp,
 311         &blocksize, &file_num, &blockno)) {
 312         set_error_from_ndmp(self);
 313         return FALSE;
 314     }
 315
 316     g_assert(file_num < INT_MAX);
 317     dself->file = (int)file_num;
 318     dself->block = blockno;
 319
 320     return TRUE;
 321 }
 322
 323 static robust_write_result
 324 robust_write(
 325     NdmpDevice *self,
 326     char *buf,
 327     guint64 count)
 328 {
 329     guint64 actual;
 330     robust_write_result subresult;
 331
 332     if (!ndmp_connection_tape_write(self->ndmp, buf, count, &actual)) {
 333         switch (ndmp_connection_err_code(self->ndmp)) {
 334             case NDMP9_IO_ERR:
 335                 /* We encountered PEOM; this only happens when the caller ignores
 336                  * LEOM */
 337                 return ROBUST_WRITE_NO_SPACE;
 338
 339             case NDMP9_EOM_ERR:
 340                 /* We encountered LEOM; retry the write (which should succeed) */
 341                 subresult = robust_write(self, buf, count);
 342                 if (subresult != ROBUST_WRITE_OK)
 343                     return subresult;
 344                 g_debug("ndmp device hit logical EOM");
 345                 return ROBUST_WRITE_OK_LEOM;
 346
 347             default:
 348                 set_error_from_ndmp(self);
 349                 return ROBUST_WRITE_ERROR;
 350         }
 351     }
 352
 353     g_assert(count == actual);
 354     return ROBUST_WRITE_OK;
 355 }
 356
 357 static void
 358 set_error_from_ndmp(
 359     NdmpDevice *self)
 360 {
 361     /* translate some error codes to the corresponding Device API status */
 362     switch (ndmp_connection_err_code(self->ndmp)) {
 363         case NDMP9_NO_TAPE_LOADED_ERR:
 364             device_set_error(DEVICE(self),
 365                     g_strdup(_("no tape loaded")),
 366                             DEVICE_STATUS_VOLUME_MISSING);
 367             break;
 368
 369         case NDMP9_DEVICE_BUSY_ERR:
 370             device_set_error(DEVICE(self),
 371                     g_strdup(_("device busy")),
 372                             DEVICE_STATUS_DEVICE_BUSY);
 373             break;
 374
 375         case NDMP9_IO_ERR:
 376             device_set_error(DEVICE(self),
 377                     g_strdup(_("IO error")),
 378                             DEVICE_STATUS_VOLUME_UNLABELED |
 379                             DEVICE_STATUS_VOLUME_ERROR |
 380                             DEVICE_STATUS_DEVICE_ERROR);
 381             break;
 382
 383         default:
 384             device_set_error(DEVICE(self),
 385                     ndmp_connection_err_msg(self->ndmp),
 386                     DEVICE_STATUS_DEVICE_ERROR);
 387             break;
 388         }
 389     close_connection(self);
 390 }
 391
 392 /*
 393  * Virtual function overrides
 394  */
 395
 396 static void
 397 ndmp_device_open_device(
 398     Device *dself,
 399     char   *device_name,
 400     char   *device_type,
 401     char   *device_node)
 402 {
 403     NdmpDevice          *self = NDMP_DEVICE(dself);
 404     char *colon, *at;
 405
 406     /* first, extract the various parts of the device_node:
 407      * HOST[:PORT]@DEVICE */
 408     colon = strchr(device_node, ':');
 409     at = strchr(device_node, '@');
 410     if (colon > at)
 411         colon = NULL; /* :PORT only counts if it's before the device name */
 412     if (!at) {
 413         device_set_error(dself,
 414                          g_strdup_printf("invalid ndmp device name '%s'", device_name),
 415                          DEVICE_STATUS_DEVICE_ERROR);
 416         return;
 417     }
 418
 419     if (colon) {
 420         char *p = NULL;
 421         long port = strtol(colon+1, &p, 10);
 422
 423         if (port < 0 || port >= 65536 || p != at || (!port && EINVAL == errno)) {
 424             device_set_error(dself,
 425                             g_strdup_printf("invalid ndmp port in device name '%s'",
 426                                             device_name),
 427                             DEVICE_STATUS_DEVICE_ERROR);
 428             return;
 429         }
 430         self->ndmp_port = (gint)port;
 431         self->ndmp_hostname = g_strndup(device_node, colon-device_node);
 432     } else {
 433         self->ndmp_port = 0; /* (use ndmjob's default, 10000) */
 434         self->ndmp_hostname = g_strndup(device_node, at-device_node);
 435     }
 436     self->ndmp_device_name = g_strdup(at+1);
 437
 438     if (parent_class->open_device) {
 439         parent_class->open_device(dself, device_name, device_type, device_node);
 440     }
 441 }
 442
 443 static void ndmp_device_finalize(GObject * obj_self)
 444 {
 445     NdmpDevice       *self = NDMP_DEVICE (obj_self);
 446
 447     if(G_OBJECT_CLASS(parent_class)->finalize)
 448         (* G_OBJECT_CLASS(parent_class)->finalize)(obj_self);
 449
 450     (void)close_tape_agent(self); /* ignore any error */
 451
 452     if (self->directtcp_conn)
 453         g_object_unref(self->directtcp_conn);
 454
 455     if (self->listen_addrs)
 456         g_free(self->listen_addrs);
 457
 458     close_connection(self);
 459
 460     if (self->ndmp_hostname)
 461         g_free(self->ndmp_hostname);
 462     if (self->ndmp_device_name)
 463         g_free(self->ndmp_device_name);
 464     if (self->ndmp_username)
 465         g_free(self->ndmp_username);
 466     if (self->ndmp_password)
 467         g_free(self->ndmp_password);
 468     if (self->ndmp_auth)
 469         g_free(self->ndmp_auth);
 470     if (self->indirecttcp_sock != -1)
 471         close(self->indirecttcp_sock);
 472 }
 473
 474 static DeviceStatusFlags
 475 ndmp_device_read_label(
 476     Device *dself)
 477 {
 478     NdmpDevice       *self = NDMP_DEVICE(dself);
 479     dumpfile_t       *header = NULL;
 480     gpointer buf = NULL;
 481     guint64 buf_size = 0;
 482     gsize read_block_size = 0;
 483
 484     amfree(dself->volume_label);
 485     amfree(dself->volume_time);
 486     dumpfile_free(dself->volume_header);
 487     dself->volume_header = NULL;
 488
 489     if (device_in_error(self)) return dself->status;
 490
 491     if (!open_tape_agent(self)) {
 492         /* error status was set by open_tape_agent */
 493         return dself->status;
 494     }
 495
 496     if (!single_ndmp_mtio(self, NDMP9_MTIO_REW)) {
 497         /* error message, if any, is set by single_ndmp_mtio */
 498         return dself->status;
 499     }
 500
 501     /* read the tape header from the NDMP server */
 502     dself->status = 0;
 503     read_block_size = ndmp_device_read_size(self);
 504     buf = g_malloc(read_block_size);
 505     if (!ndmp_connection_tape_read(self->ndmp,
 506         buf,
 507         read_block_size,
 508         &buf_size)) {
 509
 510         /* handle known errors */
 511         switch (ndmp_connection_err_code(self->ndmp)) {
 512             case NDMP9_NO_TAPE_LOADED_ERR:
 513                 device_set_error(dself,
 514                         g_strdup(_("no tape loaded")),
 515                                 DEVICE_STATUS_VOLUME_MISSING);
 516                 goto read_err;
 517
 518             case NDMP9_IO_ERR:
 519                 device_set_error(dself,
 520                         g_strdup(_("IO error reading tape label")),
 521                                 DEVICE_STATUS_VOLUME_UNLABELED |
 522                                 DEVICE_STATUS_VOLUME_ERROR |
 523                                 DEVICE_STATUS_DEVICE_ERROR);
 524                 goto read_err;
 525
 526             case NDMP9_EOM_ERR:
 527             case NDMP9_EOF_ERR:
 528                 device_set_error(dself,
 529                         g_strdup(_("no tape label found")),
 530                                 DEVICE_STATUS_VOLUME_UNLABELED);
 531                 header = dself->volume_header = g_new(dumpfile_t, 1);
 532                 fh_init(header);
 533                 goto read_err;
 534
 535             default:
 536                 set_error_from_ndmp(self);
 537                 goto read_err;
 538             }
 539         }
 540
 541         header = dself->volume_header = g_new(dumpfile_t, 1);
 542         fh_init(header);
 543         parse_file_header(buf, header, buf_size);
 544
 545 read_err:
 546     g_free(buf);
 547
 548     if (dself->status != 0) {
 549         /* error already set above */
 550         return dself->status;
 551     }
 552
 553     /* handle a "weird" label */
 554     if (header->type != F_TAPESTART) {
 555         device_set_error(dself,
 556                 stralloc(_("No tapestart header -- unlabeled device?")),
 557                          DEVICE_STATUS_VOLUME_UNLABELED);
 558         return dself->status;
 559     }
 560     dself->volume_label = g_strdup(header->name);
 561     dself->volume_time = g_strdup(header->datestamp);
 562     /* dself->volume_header is already set */
 563
 564     /* note: connection is left open, as well as the tape device */
 565
 566     device_set_error(dself, NULL, DEVICE_STATUS_SUCCESS);
 567
 568     return dself->status;
 569 }
 570
 571
 572 static gboolean
 573 ndmp_device_start(
 574     Device           *dself,
 575     DeviceAccessMode  mode,
 576     char             *label,
 577     char             *timestamp)
 578 {
 579     NdmpDevice *self = NDMP_DEVICE(dself);
 580     dumpfile_t *header;
 581     char       *header_buf;
 582
 583     if (device_in_error(self)) return FALSE;
 584
 585     if (!open_tape_agent(self)) {
 586         /* error status was set by open_tape_agent */
 587         return FALSE;
 588     }
 589
 590     if (mode != ACCESS_WRITE && dself->volume_label == NULL) {
 591         if (ndmp_device_read_label(dself) != DEVICE_STATUS_SUCCESS)
 592             /* the error was set by ndmp_device_read_label */
 593             return FALSE;
 594     }
 595
 596     dself->access_mode = mode;
 597     g_mutex_lock(dself->device_mutex);
 598     dself->in_file = FALSE;
 599     g_mutex_unlock(dself->device_mutex);
 600
 601     if (!single_ndmp_mtio(self, NDMP9_MTIO_REW)) {
 602         /* single_ndmp_mtio already set our error message */
 603         return FALSE;
 604     }
 605
 606     /* Position the tape */
 607     switch (mode) {
 608     case ACCESS_APPEND:
 609         device_set_error(dself,
 610             g_strdup("operation not supported"),
 611             DEVICE_STATUS_DEVICE_ERROR);
 612         return FALSE;
 613         break;
 614
 615     case ACCESS_READ:
 616         dself->file = 0;
 617         break;
 618
 619     case ACCESS_WRITE:
 620         header = make_tapestart_header(dself, label, timestamp);
 621         g_assert(header != NULL);
 622
 623         header_buf = device_build_amanda_header(dself, header, NULL);
 624         if (header_buf == NULL) {
 625             device_set_error(dself,
 626                 stralloc(_("Tapestart header won't fit in a single block!")),
 627                 DEVICE_STATUS_DEVICE_ERROR);
 628             dumpfile_free(header);
 629             return FALSE;
 630         }
 631
 632         switch (robust_write(self, header_buf, dself->block_size)) {
 633             case ROBUST_WRITE_OK_LEOM:
 634                 dself->is_eom = TRUE;
 635                 /* fall through */
 636             case ROBUST_WRITE_OK:
 637                 break;
 638
 639             case ROBUST_WRITE_NO_SPACE:
 640                 /* this would be an odd error to see writing the tape label, but
 641                  * oh well */
 642                 device_set_error(dself,
 643                     stralloc(_("No space left on device")),
 644                     DEVICE_STATUS_VOLUME_ERROR);
 645                 dself->is_eom = TRUE;
 646                 /* fall through */
 647
 648             case ROBUST_WRITE_ERROR:
 649                 /* error was set by robust_write or above */
 650                 dumpfile_free(header);
 651                 amfree(header_buf);
 652                 return FALSE;
 653
 654         }
 655         amfree(header_buf);
 656
 657         if (!single_ndmp_mtio(self, NDMP9_MTIO_EOF)) {
 658             /* error was set by single_ndmp_mtio */
 659             dumpfile_free(header);
 660             return FALSE;
 661         }
 662
 663         dself->volume_label = newstralloc(dself->volume_label, label);
 664         dself->volume_time = newstralloc(dself->volume_time, timestamp);
 665         dumpfile_free(dself->volume_header);
 666         dself->volume_header = header;
 667
 668         /* unset the VOLUME_UNLABELED flag, if it was set */
 669         device_set_error(dself, NULL, DEVICE_STATUS_SUCCESS);
 670         dself->file = 0;
 671         break;
 672
 673     default:
 674         g_assert_not_reached();
 675     }
 676
 677     return TRUE;
 678 }
 679
 680 static gboolean
 681 ndmp_device_finish(
 682     Device *dself)
 683 {
 684     gboolean rval;
 685
 686     NdmpDevice *self = NDMP_DEVICE(dself);
 687     rval = !device_in_error(dself);
 688
 689     /* we're not in a file anymore */
 690     dself->access_mode = ACCESS_NULL;
 691
 692     if (!close_tape_agent(self)) {
 693         /* error is set by close_tape_agent */
 694         rval = FALSE;
 695     }
 696
 697     if (self->ndmp)
 698         close_connection(self);
 699
 700     return rval;
 701 }
 702
 703 static gboolean
 704 ndmp_device_eject(
 705     Device *dself)
 706 {
 707     NdmpDevice *self = NDMP_DEVICE(dself);
 708     if (device_in_error(dself)) return FALSE;
 709
 710     if (!single_ndmp_mtio(self, NDMP9_MTIO_OFF)) {
 711         /* error was set by single_ndmp_mtio */
 712         return FALSE;
 713     }
 714
 715     return TRUE;
 716 }
 717
 718
 719 /* functions for writing */
 720
 721 static gboolean
 722 ndmp_device_start_file(
 723     Device     *dself,
 724     dumpfile_t *header)
 725 {
 726     NdmpDevice *self = NDMP_DEVICE(dself);
 727     char *header_buf;
 728
 729     if (device_in_error(self)) return FALSE;
 730
 731     dself->is_eof = FALSE;
 732     dself->is_eom = FALSE;
 733     g_mutex_lock(dself->device_mutex);
 734     dself->bytes_written = 0;
 735     g_mutex_unlock(dself->device_mutex);
 736
 737     /* set the blocksize in the header properly */
 738     header->blocksize = dself->block_size;
 739
 740     header_buf = device_build_amanda_header(dself, header, NULL);
 741     if (header_buf == NULL) {
 742         device_set_error(dself,
 743             stralloc(_("Amanda file header won't fit in a single block!")),
 744             DEVICE_STATUS_DEVICE_ERROR);
 745         return FALSE;
 746     }
 747
 748     switch (robust_write(self, header_buf, dself->block_size)) {
 749         case ROBUST_WRITE_OK_LEOM:
 750             dself->is_eom = TRUE;
 751             /* fall through */
 752
 753         case ROBUST_WRITE_OK:
 754             break;
 755
 756         case ROBUST_WRITE_NO_SPACE:
 757             /* this would be an odd error to see writing the tape label, but
 758              * oh well */
 759             device_set_error(dself,
 760                 stralloc(_("No space left on device")),
 761                 DEVICE_STATUS_VOLUME_ERROR);
 762             dself->is_eom = TRUE;
 763             /* fall through */
 764
 765         case ROBUST_WRITE_ERROR:
 766             /* error was set by robust_write or above */
 767             amfree(header_buf);
 768             return FALSE;
 769     }
 770     amfree(header_buf);
 771
 772     /* arrange the file numbers correctly */
 773     g_mutex_lock(dself->device_mutex);
 774     dself->in_file = TRUE;
 775     g_mutex_unlock(dself->device_mutex);
 776     if (!ndmp_get_state(self)) {
 777         /* error already set by ndmp_get_state */
 778         return FALSE;
 779     }
 780
 781     /* double-check that the tape agent gave us a non-bogus file number */
 782     g_assert(dself->file > 0);
 783
 784     return TRUE;
 785 }
 786
 787 static gboolean
 788 ndmp_device_write_block(
 789     Device   *dself,
 790     guint     size,
 791     gpointer  data)
 792 {
 793     NdmpDevice *self = NDMP_DEVICE(dself);
 794     gpointer replacement_buffer = NULL;
 795
 796     if (device_in_error(self)) return FALSE;
 797
 798     /* zero out to the end of a short block -- tape devices only write
 799      * whole blocks. */
 800     if (size < dself->block_size) {
 801         replacement_buffer = malloc(dself->block_size);
 802         memcpy(replacement_buffer, data, size);
 803         bzero(replacement_buffer+size, dself->block_size-size);
 804
 805         data = replacement_buffer;
 806         size = dself->block_size;
 807     }
 808
 809     switch (robust_write(self, data, size)) {
 810         case ROBUST_WRITE_OK_LEOM:
 811             dself->is_eom = TRUE;
 812             /* fall through */
 813
 814         case ROBUST_WRITE_OK:
 815             break;
 816
 817         case ROBUST_WRITE_NO_SPACE:
 818             /* this would be an odd error to see writing the tape label, but
 819              * oh well */
 820             device_set_error(dself,
 821                 stralloc(_("No space left on device")),
 822                 DEVICE_STATUS_VOLUME_ERROR);
 823             dself->is_eom = TRUE;
 824             /* fall through */
 825
 826         case ROBUST_WRITE_ERROR:
 827             /* error was set by robust_write or above */
 828             if (replacement_buffer) g_free(replacement_buffer);
 829             return FALSE;
 830     }
 831
 832     dself->block++;
 833     g_mutex_lock(dself->device_mutex);
 834     dself->bytes_written += size;
 835     g_mutex_unlock(dself->device_mutex);
 836
 837     if (replacement_buffer) g_free(replacement_buffer);
 838     return TRUE;
 839 }
 840
 841 static gboolean
 842 ndmp_device_finish_file(
 843     Device *dself)
 844 {
 845     NdmpDevice *self = NDMP_DEVICE(dself);
 846
 847     if (device_in_error(dself)) return FALSE;
 848
 849     /* we're not in a file anymore */
 850     g_mutex_lock(dself->device_mutex);
 851     dself->in_file = FALSE;
 852     g_mutex_unlock(dself->device_mutex);
 853
 854     if (!single_ndmp_mtio(self, NDMP9_MTIO_EOF)) {
 855         /* error was set by single_ndmp_mtio */
 856         dself->is_eom = TRUE;
 857         return FALSE;
 858     }
 859
 860     return TRUE;
 861 }
 862
 863 /* functions for reading */
 864
 865 static dumpfile_t*
 866 ndmp_device_seek_file(
 867     Device *dself,
 868     guint   file)
 869 {
 870     NdmpDevice *self = NDMP_DEVICE(dself);
 871     gint delta;
 872     guint resid;
 873     gpointer buf;
 874     guint64 buf_size;
 875     dumpfile_t *header;
 876     gsize read_block_size = 0;
 877
 878     if (device_in_error(dself)) return FALSE;
 879
 880     /* file 0 is the tape header, and isn't seekable as a distinct
 881      * Device-API-level file */
 882     if (file == 0) {
 883         device_set_error(dself,
 884             g_strdup("cannot seek to file 0"),
 885             DEVICE_STATUS_DEVICE_ERROR);
 886         return NULL;
 887     }
 888
 889     /* first, make sure the file and block numbers are correct */
 890     if (!ndmp_get_state(self)) {
 891         /* error already set by ndmp_get_state */
 892         return FALSE;
 893     }
 894
 895     /* now calculate the file delta */
 896     delta = file - dself->file;
 897
 898     if (delta <= 0) {
 899         /* Note that this algorithm will rewind to the beginning of
 900          * the current part, too */
 901
 902         /* BSF *past* the filemark we want to seek to */
 903         if (!ndmp_connection_tape_mtio(self->ndmp, NDMP9_MTIO_BSF, -delta + 1, &resid)) {
 904             set_error_from_ndmp(self);
 905             return NULL;
 906         }
 907         if (resid != 0)
 908             goto incomplete_bsf;
 909
 910         /* now we are on the BOT side of the filemark, but we want to be
 911          * on the EOT side of it.  An FSF will get us there.. */
 912         if (!ndmp_connection_tape_mtio(self->ndmp, NDMP9_MTIO_FSF, 1, &resid)) {
 913             set_error_from_ndmp(self);
 914             return NULL;
 915         }
 916
 917         if (resid != 0) {
 918 incomplete_bsf:
 919             device_set_error(dself,
 920                 g_strdup_printf("BSF operation failed to seek by %d files", resid),
 921                 DEVICE_STATUS_DEVICE_ERROR);
 922             return NULL;
 923         }
 924     } else /* (delta > 0) */ {
 925         if (!ndmp_connection_tape_mtio(self->ndmp, NDMP9_MTIO_FSF, delta, &resid)) {
 926             set_error_from_ndmp(self);
 927             return FALSE;
 928         }
 929
 930         /* if we didn't seek all the way there, then we're past the tapeend */
 931         if (resid > 0) {
 932             device_set_error(dself,
 933                 vstrallocf(_("Could not seek forward to file %d"), file),
 934                 DEVICE_STATUS_VOLUME_ERROR);
 935             return NULL;
 936         }
 937     }
 938
 939     /* fix up status */
 940     g_mutex_lock(dself->device_mutex);
 941     dself->in_file = TRUE;
 942     g_mutex_unlock(dself->device_mutex);
 943     dself->file = file;
 944     dself->block = 0;
 945     g_mutex_lock(dself->device_mutex);
 946     dself->bytes_read = 0;
 947     g_mutex_unlock(dself->device_mutex);
 948
 949     /* now read the header */
 950     read_block_size = ndmp_device_read_size(self);
 951     buf = g_malloc(read_block_size);
 952     if (!ndmp_connection_tape_read(self->ndmp,
 953                 buf, read_block_size, &buf_size)) {
 954         switch (ndmp_connection_err_code(self->ndmp)) {
 955             case NDMP9_EOF_ERR:
 956             case NDMP9_EOM_ERR:
 957                 return make_tapeend_header();
 958
 959             default:
 960                 set_error_from_ndmp(self);
 961                 g_free(buf);
 962                 return NULL;
 963         }
 964     }
 965
 966     header = g_new(dumpfile_t, 1);
 967     fh_init(header);
 968     parse_file_header(buf, header, buf_size);
 969     g_free(buf);
 970
 971     return header;
 972 }
 973
 974 static gboolean
 975 ndmp_device_seek_block(
 976     Device  *dself,
 977     guint64  block)
 978 {
 979     if (device_in_error(dself)) return FALSE;
 980
 981     dself->block = block;
 982
 983     device_set_error(dself, g_strdup("operation not supported"), DEVICE_STATUS_DEVICE_ERROR);
 984     return FALSE;
 985 }
 986
 987 static int
 988 ndmp_device_read_block (Device * dself, gpointer data, int *size_req) {
 989     NdmpDevice *self = NDMP_DEVICE(dself);
 990     guint64 requested, actual;
 991     gsize read_block_size = ndmp_device_read_size(self);
 992
 993     /* We checked the NDMP device's blocksize when the device was opened, which should
 994      * catch any misalignent of server block size and Amanda block size */
 995
 996     g_assert(read_block_size < INT_MAX); /* check data type mismatch */
 997     if (!data || *size_req < (int)(read_block_size)) {
 998         *size_req = (int)(read_block_size);
 999         return 0;
1000     }
1001
1002     requested = *size_req;
1003     if (!ndmp_connection_tape_read(self->ndmp,
1004         data,
1005         requested,
1006         &actual)) {
1007
1008         /* handle known errors */
1009         switch (ndmp_connection_err_code(self->ndmp)) {
1010             case NDMP9_EOM_ERR:
1011             case NDMP9_EOF_ERR:
1012                 dself->is_eof = TRUE;
1013                 return -1;
1014
1015             default:
1016                 set_error_from_ndmp(self);
1017                 return -1;
1018         }
1019     }
1020
1021     *size_req = (int)actual; /* cast is OK - requested size was < INT_MAX too */
1022     g_mutex_lock(dself->device_mutex);
1023     dself->bytes_read += actual;
1024     g_mutex_unlock(dself->device_mutex);
1025
1026     return *size_req;
1027 }
1028
1029 static gboolean
1030 indirecttcp_listen(
1031     NdmpDevice *self,
1032     DirectTCPAddr **addrs)
1033 {
1034     in_port_t port;
1035
1036     self->indirecttcp_sock = stream_server(AF_INET, &port, 0, STREAM_BUFSIZE, 0);
1037     if (self->indirecttcp_sock < 0) {
1038         device_set_error(DEVICE(self),
1039             g_strdup_printf("Could not bind indirecttcp socket: %s", strerror(errno)),
1040             DEVICE_STATUS_DEVICE_ERROR);
1041         return FALSE;
1042     }
1043
1044     /* An IndirectTCP address is 255.255.255.255:$port */
1045     self->listen_addrs = *addrs = g_new0(DirectTCPAddr, 2);
1046     addrs[0]->sin.sin_family = AF_INET;
1047     addrs[0]->sin.sin_addr.s_addr = htonl(0xffffffff);
1048     SU_SET_PORT(addrs[0], port);
1049
1050     return TRUE;
1051 }
1052
1053 static gboolean
1054 listen_impl(
1055     Device *dself,
1056     gboolean for_writing,
1057     DirectTCPAddr **addrs)
1058 {
1059     NdmpDevice *self = NDMP_DEVICE(dself);
1060
1061     if (device_in_error(self)) return FALSE;
1062
1063     /* check status */
1064     g_assert(!self->listen_addrs);
1065
1066     if (!open_tape_agent(self)) {
1067         /* error message was set by open_tape_agent */
1068         return FALSE;
1069     }
1070
1071     self->for_writing = for_writing;
1072
1073     /* first, set the window to an empty span so that the mover doesn't start
1074      * reading or writing data immediately.  NDMJOB tends to reset the record
1075      * size periodically (in direct contradiction to the spec), so we reset it
1076      * here as well. */
1077     if (!ndmp_connection_mover_set_record_size(self->ndmp,
1078                 DEVICE(self)->block_size)) {
1079         set_error_from_ndmp(self);
1080         return FALSE;
1081     }
1082
1083     if (for_writing) {
1084         /* if we're forcing indirecttcp, just do it */
1085         if (self->indirect) {
1086             return indirecttcp_listen(self, addrs);
1087         }
1088         if (!ndmp_connection_mover_set_window(self->ndmp, 0, 0)) {
1089             /* NDMP4_ILLEGAL_ARGS_ERR means the NDMP server doesn't like a zero-byte
1090              * mover window, so we'll ignore it */
1091             if (ndmp_connection_err_code(self->ndmp) != NDMP4_ILLEGAL_ARGS_ERR) {
1092                 set_error_from_ndmp(self);
1093                 return FALSE;
1094             }
1095
1096             g_debug("NDMP Device: cannot set zero-length mover window; "
1097                     "falling back to IndirectTCP");
1098             /* In this case, we need to set up IndirectTCP */
1099             return indirecttcp_listen(self, addrs);
1100         }
1101     } else {
1102         /* For reading, set the window to the second mover record, so that the
1103          * mover will pause immediately when it wants to read the first mover
1104          * record. */
1105         if (!ndmp_connection_mover_set_window(self->ndmp,
1106                                               dself->block_size,
1107                                               dself->block_size)) {
1108             set_error_from_ndmp(self);
1109             return FALSE;
1110         }
1111     }
1112
1113     /* then tell it to start listening */
1114     if (!ndmp_connection_mover_listen(self->ndmp,
1115                 for_writing? NDMP9_MOVER_MODE_READ : NDMP9_MOVER_MODE_WRITE,
1116                 NDMP9_ADDR_TCP,
1117                 addrs)) {
1118         set_error_from_ndmp(self);
1119         return FALSE;
1120     }
1121     self->listen_addrs = *addrs;
1122
1123     return TRUE;
1124 }
1125
1126 static gpointer
1127 accept_wait_cond(
1128     gpointer data)
1129 {
1130     NdmpDevice *self = NDMP_DEVICE(data);
1131
1132     ndmp9_mover_state state;
1133     guint64 bytes_moved;
1134
1135     gulong backoff = G_USEC_PER_SEC/20; /* 5 msec */
1136     g_mutex_lock(self->abort_mutex);
1137     while (1) {
1138         g_mutex_unlock(self->abort_mutex);
1139         if (!ndmp_connection_mover_get_state(self->ndmp,
1140                     &state, &bytes_moved, NULL, NULL)) {
1141             g_mutex_lock(self->abort_mutex);
1142             set_error_from_ndmp(self);
1143             state = 0;
1144             break;
1145         }
1146         g_mutex_lock(self->abort_mutex);
1147
1148         if (state != NDMP9_MOVER_STATE_LISTEN)
1149             break;
1150
1151         /* back off a little bit to give the other side time to breathe,
1152          * but not more than one second */
1153         g_mutex_unlock(self->abort_mutex);
1154         g_usleep(backoff);
1155         g_mutex_lock(self->abort_mutex);
1156         if (self->cancel)
1157             break;
1158         backoff *= 2;
1159         if (backoff > G_USEC_PER_SEC)
1160             backoff = G_USEC_PER_SEC;
1161     }
1162
1163     self->cancel = TRUE;
1164     g_cond_broadcast(self->abort_cond);
1165     g_mutex_unlock(self->abort_mutex);
1166     return GINT_TO_POINTER(state);
1167 }
1168
1169 static int
1170 accept_impl(
1171     Device *dself,
1172     DirectTCPConnection **dtcpconn,
1173     gboolean *cancelled,
1174     GMutex *abort_mutex,
1175     GCond *abort_cond)
1176 {
1177     NdmpDevice *self = NDMP_DEVICE(dself);
1178     ndmp9_mover_state state;
1179     ndmp9_mover_mode mode;
1180     ndmp9_mover_halt_reason mover_halt_reason  = NDMP9_MOVER_HALT_NA;
1181     ndmp9_mover_pause_reason mover_pause_reason = NDMP9_MOVER_PAUSE_NA;
1182     guint64 seek_position;
1183     GThread *wait_thread;
1184     int      result;
1185     char    *err;
1186
1187     if (device_in_error(self)) return 1;
1188
1189     self->abort_mutex = abort_mutex;
1190     self->abort_cond = abort_cond;
1191     self->cancelled = cancelled;
1192     self->cancel = FALSE;
1193
1194     g_assert(self->listen_addrs);
1195
1196     *dtcpconn = NULL;
1197
1198     if (!self->for_writing) {
1199         /* when reading, we don't get any kind of notification that the
1200          * connection has been established, but we can't call NDMP_MOVER_READ
1201          * until the mover is active.  So we have to poll, waiting for ACTIVE.
1202          * This is ugly. */
1203
1204         wait_thread = g_thread_create(accept_wait_cond, (gpointer)self, TRUE,
1205                                       NULL);
1206         while (!*cancelled && !self->cancel) {
1207             g_cond_wait(self->abort_cond, self->abort_mutex);
1208         }
1209         self->cancel = TRUE;
1210         state = GPOINTER_TO_INT(g_thread_join(wait_thread));
1211
1212         if (*cancelled) {
1213             result = 2;
1214             goto accept_failed;
1215         }
1216
1217         /* double-check state */
1218         if (state != NDMP9_MOVER_STATE_ACTIVE) {
1219             device_set_error(DEVICE(self),
1220                 g_strdup("mover did not enter the ACTIVE state as expected"),
1221                 DEVICE_STATUS_DEVICE_ERROR);
1222             result = 1;
1223             goto accept_failed;
1224         }
1225
1226         /* now, we need to get this into the PAUSED state, since right now we
1227          * aren't allowed to perform any tape movement commands.  So we issue a
1228          * MOVER_READ request for the whole darn image stream after setting the
1229          * usual empty window. Note that this means the whole dump will be read
1230          * in one MOVER_READ operation, even if it does not begin at the
1231          * beginning of a part. */
1232         if (!ndmp_connection_mover_read(self->ndmp, 0, G_MAXUINT64)) {
1233             set_error_from_ndmp(self);
1234             result = 1;
1235             goto accept_failed;
1236         }
1237
1238         /* now we should expect a notice that the mover has paused */
1239     } else {
1240         /* when writing, the mover will pause as soon as the first byte comes
1241          * in, so there's no need to do anything to trigger the pause. */
1242     }
1243
1244     if (self->indirecttcp_sock == -1) {
1245         /* NDMJOB sends NDMP9_MOVER_PAUSE_SEEK to indicate that it wants to
1246          * write outside the window, while the standard specifies .._EOW,
1247          * instead.  When reading to a connection, we get the appropriate
1248          * .._SEEK.  It's easy enough to handle both. */
1249         result = ndmp_connection_wait_for_notify_with_cond(self->ndmp,
1250                         NULL,
1251                         &mover_halt_reason,
1252                         &mover_pause_reason, &seek_position,
1253                         cancelled,
1254                         abort_mutex, abort_cond);
1255
1256         if (result == 1) {
1257             set_error_from_ndmp(self);
1258             goto accept_failed;
1259         } else if (result == 2) {
1260             goto accept_failed;
1261         }
1262
1263         err = NULL;
1264         if (mover_pause_reason) {
1265             switch (mover_pause_reason) {
1266                 case NDMP9_MOVER_PAUSE_SEEK:
1267                 case NDMP9_MOVER_PAUSE_EOW:
1268                         break;
1269                 default:
1270                         err = "got NOTIFY_MOVER_PAUSED, but not because of EOW or SEEK";
1271                         break;
1272             }
1273         } else if (mover_halt_reason) {
1274             err = "unexpected NOTIFY_MOVER_HALT";
1275         }
1276
1277         if (err) {
1278             device_set_error(DEVICE(self),
1279                 g_strdup_printf("waiting NDMP_MOVER_PAUSE_SEEK: %s", err),
1280                 DEVICE_STATUS_DEVICE_ERROR);
1281             result = 1;
1282             goto accept_failed;
1283
1284         }
1285     }
1286
1287     /* NDMJOB sends NDMP9_MOVER_PAUSE_SEEK to indicate that it wants to write
1288      * outside the window, while the standard specifies .._EOW, instead.  When
1289      * reading to a connection, we get the appropriate .._SEEK.  It's easy
1290      * enough to handle both. */
1291
1292     if (self->indirecttcp_sock == -1) {
1293         g_free(self->listen_addrs);
1294         self->listen_addrs = NULL;
1295     }
1296
1297     if (self->for_writing)
1298         mode = NDMP9_MOVER_MODE_READ;
1299     else
1300         mode = NDMP9_MOVER_MODE_WRITE;
1301
1302     /* set up the new directtcp connection */
1303     if (self->directtcp_conn)
1304         g_object_unref(self->directtcp_conn);
1305     self->directtcp_conn =
1306         directtcp_connection_ndmp_new(self->ndmp, mode);
1307     *dtcpconn = DIRECTTCP_CONNECTION(self->directtcp_conn);
1308
1309     /* reference it for the caller */
1310     g_object_ref(*dtcpconn);
1311
1312     return 0;
1313
1314 accept_failed:
1315     if (self->indirecttcp_sock == -1) {
1316         g_free(self->listen_addrs);
1317         self->listen_addrs = NULL;
1318     }
1319     return result;
1320 }
1321
1322
1323 static int
1324 connect_impl(
1325     Device *dself,
1326     gboolean for_writing,
1327     DirectTCPAddr *addrs,
1328     DirectTCPConnection **dtcpconn,
1329     int    *cancelled,
1330     GMutex *abort_mutex,
1331     GCond  *abort_cond)
1332 {
1333     NdmpDevice *self = NDMP_DEVICE(dself);
1334     ndmp9_mover_mode mode;
1335     ndmp9_mover_halt_reason mover_halt_reason = NDMP9_MOVER_HALT_NA;
1336     ndmp9_mover_pause_reason mover_pause_reason = NDMP9_MOVER_PAUSE_NA;
1337     guint64 seek_position;
1338     int result;
1339
1340     g_assert(!self->listen_addrs);
1341
1342     *dtcpconn = NULL;
1343     self->for_writing = for_writing;
1344
1345     if (!open_tape_agent(self)) {
1346         /* error message was set by open_tape_agent */
1347         return 1;
1348     }
1349
1350     /* first, set the window to an empty span so that the mover doesn't start
1351      * reading or writing data immediately.  NDMJOB tends to reset the record
1352      * size periodically (in direct contradiction to the spec), so we reset it
1353      * here as well. */
1354     if (!ndmp_connection_mover_set_record_size(self->ndmp,
1355                 DEVICE(self)->block_size)) {
1356         set_error_from_ndmp(self);
1357         return 1;
1358     }
1359
1360     if (!ndmp_connection_mover_set_window(self->ndmp, 0, 0)) {
1361         set_error_from_ndmp(self);
1362         return 1;
1363     }
1364
1365     if (self->for_writing)
1366         mode = NDMP9_MOVER_MODE_READ;
1367     else
1368         mode = NDMP9_MOVER_MODE_WRITE;
1369
1370     if (!ndmp_connection_mover_connect(self->ndmp, mode, addrs)) {
1371         set_error_from_ndmp(self);
1372         return 1;
1373     }
1374
1375     if (!self->for_writing) {
1376         /* The agent is in the ACTIVE state, and will remain so until we tell
1377          * it to do something else.  The thing we want to is for it to start
1378          * reading data from the tape, which will immediately trigger an EOW or
1379          * SEEK pause. */
1380         if (!ndmp_connection_mover_read(self->ndmp, 0, G_MAXUINT64)) {
1381             set_error_from_ndmp(self);
1382             return 1;
1383         }
1384
1385         /* now we should expect a notice that the mover has paused */
1386     } else {
1387         /* when writing, the mover will pause as soon as the first byte comes
1388          * in, so there's no need to do anything to trigger the pause. */
1389     }
1390
1391     /* NDMJOB sends NDMP9_MOVER_PAUSE_SEEK to indicate that it wants to write
1392      * outside the window, while the standard specifies .._EOW, instead.  When
1393      * reading to a connection, we get the appropriate .._SEEK.  It's easy
1394      * enough to handle both. */
1395
1396     result = ndmp_connection_wait_for_notify_with_cond(self->ndmp,
1397             NULL,
1398             &mover_halt_reason,
1399             &mover_pause_reason, &seek_position,
1400             cancelled,
1401             abort_mutex, abort_cond);
1402
1403     if (result == 1) {
1404         set_error_from_ndmp(self);
1405         return 1;
1406     } else if (result == 2) {
1407         return 2;
1408     }
1409
1410     if (mover_halt_reason != NDMP9_MOVER_HALT_NA) {
1411         device_set_error(DEVICE(self),
1412             g_strdup_printf("got NDMP9_MOVER_HALT"),
1413             DEVICE_STATUS_DEVICE_ERROR);
1414         return 1;
1415     }
1416     if (mover_pause_reason != NDMP9_MOVER_PAUSE_SEEK &&
1417         mover_pause_reason != NDMP9_MOVER_PAUSE_EOW) {
1418         device_set_error(DEVICE(self),
1419             g_strdup_printf("got NOTIFY_MOVER_PAUSED, but not because of EOW or SEEK"),
1420             DEVICE_STATUS_DEVICE_ERROR);
1421         return 1;
1422     }
1423
1424     if (self->listen_addrs) {
1425         g_free(self->listen_addrs);
1426         self->listen_addrs = NULL;
1427     }
1428
1429     /* set up the new directtcp connection */
1430     if (self->directtcp_conn)
1431         g_object_unref(self->directtcp_conn);
1432     self->directtcp_conn =
1433         directtcp_connection_ndmp_new(self->ndmp, mode);
1434     *dtcpconn = DIRECTTCP_CONNECTION(self->directtcp_conn);
1435
1436     /* reference it for the caller */
1437     g_object_ref(*dtcpconn);
1438
1439     return 0;
1440 }
1441
1442 static gboolean
1443 indirecttcp_start_writing(
1444         NdmpDevice *self)
1445 {
1446     DirectTCPAddr *real_addrs, *iter;
1447     int conn_sock;
1448
1449     /* The current state is that the other end is trying to connect to
1450      * indirecttcp_sock.  The mover remains IDLE, although its window is set
1451      * correctly for the part we are about to write. */
1452
1453     g_debug("indirecttcp_start_writing, ready to accept");
1454     conn_sock = accept(self->indirecttcp_sock, NULL, NULL);
1455     if (conn_sock < 0) {
1456         device_set_error(DEVICE(self),
1457             g_strdup_printf("Could not accept indirecttcp socket: %s", strerror(errno)),
1458             DEVICE_STATUS_DEVICE_ERROR);
1459         return FALSE;
1460     }
1461     g_debug("indirecttcp_start_writing, accepted");
1462
1463     close(self->indirecttcp_sock);
1464     self->indirecttcp_sock = -1;
1465
1466     /* tell mover to start listening */
1467     g_assert(self->for_writing);
1468     if (!ndmp_connection_mover_listen(self->ndmp,
1469                 NDMP4_MOVER_MODE_READ,
1470                 NDMP4_ADDR_TCP,
1471                 &real_addrs)) {
1472         set_error_from_ndmp(self);
1473         return FALSE;
1474     }
1475
1476     /* format the addresses and send them down the socket */
1477     for (iter = real_addrs; iter && SU_GET_FAMILY(iter) != 0; iter++) {
1478         char inet[INET_ADDRSTRLEN];
1479         const char *addr;
1480         char *addrspec;
1481
1482         addr = inet_ntop(AF_INET, &iter->sin.sin_addr.s_addr, inet, INET_ADDRSTRLEN);
1483
1484         addrspec = g_strdup_printf("%s:%d%s", addr, SU_GET_PORT(iter),
1485                 SU_GET_FAMILY(iter+1) !=0? " ":"");
1486         g_debug("indirecttcp_start_writing, send %s", addrspec);
1487         if (full_write(conn_sock, addrspec, strlen(addrspec)) < strlen(addrspec)) {
1488             device_set_error(DEVICE(self),
1489                 g_strdup_printf("writing to indirecttcp socket: %s", strerror(errno)),
1490                 DEVICE_STATUS_DEVICE_ERROR);
1491             return FALSE;
1492         }
1493     }
1494
1495     /* close the socket for good.  This ensures that the next call to
1496      * write_from_connection_impl will not go through the mover setup process.
1497      * */
1498     if (close(conn_sock) < 0) {
1499         device_set_error(DEVICE(self),
1500             g_strdup_printf("closing indirecttcp socket: %s", strerror(errno)),
1501             DEVICE_STATUS_DEVICE_ERROR);
1502         return FALSE;
1503     }
1504     conn_sock = -1;
1505
1506     /* and free the listen_addrs, since we didn't free them in accept_impl */
1507     if (self->listen_addrs) {
1508         g_free(self->listen_addrs);
1509         self->listen_addrs = NULL;
1510     }
1511
1512     /* Now it's up to the remote end to connect to the mover and start sending
1513      * data.  We won't get any notification when this happens, although we could
1514      * in principle poll for such a thing. */
1515     return TRUE;
1516 }
1517
1518 static int
1519 write_from_connection_impl(
1520     Device  *dself,
1521     guint64  size,
1522     guint64 *actual_size,
1523     int     *cancelled,
1524     GMutex  *abort_mutex,
1525     GCond   *abort_cond)
1526 {
1527     NdmpDevice *self = NDMP_DEVICE(dself);
1528     DirectTCPConnectionNDMP *nconn = self->directtcp_conn;
1529     gboolean eom = FALSE, eof = FALSE, eow = FALSE;
1530     ndmp9_mover_state mover_state;
1531     ndmp9_mover_halt_reason mover_halt_reason = NDMP9_MOVER_HALT_NA;
1532     ndmp9_mover_pause_reason mover_pause_reason = NDMP9_MOVER_PAUSE_NA;
1533     guint64 bytes_moved_before, bytes_moved_after;
1534     gchar *err;
1535     int result;
1536
1537     if (device_in_error(self)) return FALSE;
1538
1539     g_debug("write_from_connection_impl");
1540     if (actual_size)
1541         *actual_size = 0;
1542
1543     /* if this is false, then the caller did not use use_connection correctly */
1544     g_assert(self->directtcp_conn != NULL);
1545     g_assert(self->ndmp == nconn->ndmp);
1546     g_assert(nconn->mode == NDMP9_MOVER_MODE_READ);
1547
1548     if (!ndmp_connection_mover_get_state(self->ndmp,
1549                 &mover_state, &bytes_moved_before, NULL, NULL)) {
1550         set_error_from_ndmp(self);
1551         return 1;
1552     }
1553
1554     if (self->indirecttcp_sock != -1) {
1555         /* If we're doing IndirectTCP, then we've deferred the whole
1556          * mover_set_window mover_listen process.. until now.
1557          * So the mover should be IDLE.
1558          */
1559         g_assert(mover_state == NDMP9_MOVER_STATE_IDLE);
1560     } else {
1561         /* the mover had best be PAUSED right now */
1562         g_assert(mover_state == NDMP9_MOVER_STATE_PAUSED);
1563     }
1564
1565     /* we want to set the window regardless of whether this is directtcp or
1566      * indirecttcp
1567      */
1568     if (!ndmp_connection_mover_set_window(self->ndmp,
1569                 nconn->offset,
1570                 size? size : G_MAXUINT64 - nconn->offset)) {
1571         set_error_from_ndmp(self);
1572         return 1;
1573     }
1574
1575     /* for DirectTCP, we just tell the mover to continue; IndirectTCP is more complicated. */
1576     if (self->indirecttcp_sock != -1) {
1577         if (!indirecttcp_start_writing(self)) {
1578             return 1;
1579         }
1580     } else {
1581         if (!ndmp_connection_mover_continue(self->ndmp)) {
1582             set_error_from_ndmp(self);
1583             return 1;
1584         }
1585     }
1586
1587     /* now wait for the mover to pause itself again, or halt on EOF or an error */
1588      result = ndmp_connection_wait_for_notify_with_cond(self->ndmp,
1589                     NULL,
1590                     &mover_halt_reason,
1591                     &mover_pause_reason, NULL,
1592                     cancelled, abort_mutex, abort_cond);
1593     if (result == 1) {
1594         set_error_from_ndmp(self);
1595         return 1;
1596     } else if (result == 2) {
1597         return 2;
1598     }
1599
1600     err = NULL;
1601     if (mover_pause_reason) {
1602         switch (mover_pause_reason) {
1603             case NDMP9_MOVER_PAUSE_EOM:
1604                 eom = TRUE;
1605                 break;
1606
1607             /* ndmjob sends .._SEEK when it should send .._EOW, so deal with
1608                 * both equivalently */
1609             case NDMP9_MOVER_PAUSE_EOW:
1610             case NDMP9_MOVER_PAUSE_SEEK:
1611                 eow = TRUE;
1612                 break;
1613
1614             default:
1615                 err = "got NOTIFY_MOVER_PAUSED, but not because of EOW or SEEK";
1616                 break;
1617         }
1618     } else if (mover_halt_reason) {
1619         switch (mover_halt_reason) {
1620             case NDMP9_MOVER_HALT_CONNECT_CLOSED:
1621                 eof = TRUE;
1622                 break;
1623
1624             default:
1625             case NDMP9_MOVER_HALT_ABORTED:
1626             /* case NDMP9_MOVER_HALT_MEDIA_ERROR: <-- not in ndmjob */
1627             case NDMP9_MOVER_HALT_INTERNAL_ERROR:
1628             case NDMP9_MOVER_HALT_CONNECT_ERROR:
1629                 err = "unexpected NDMP_NOTIFY_MOVER_HALTED";
1630                 break;
1631         }
1632     }
1633
1634     if (err) {
1635         device_set_error(DEVICE(self),
1636             g_strdup_printf("waiting for accept: %s", err),
1637             DEVICE_STATUS_DEVICE_ERROR);
1638         return 1;
1639     }
1640
1641     /* no error, so the mover stopped due to one of EOM (volume out of space),
1642      * EOF (data connection is done), or EOW (maximum part size was written).
1643      * In any case, we want to know how many bytes were written. */
1644
1645     if (!ndmp_connection_mover_get_state(self->ndmp,
1646                 &mover_state, &bytes_moved_after, NULL, NULL)) {
1647         set_error_from_ndmp(self);
1648         return 1;
1649     }
1650     size = bytes_moved_after - bytes_moved_before;
1651     nconn->offset += size;
1652
1653     if (actual_size) {
1654         *actual_size = bytes_moved_after - bytes_moved_before;
1655     }
1656
1657     if (eow) {
1658         ; /* mover finished the whole part -- nothing to report! */
1659     } else if (eof) {
1660         DEVICE(self)->is_eof = TRUE;
1661     } else if (eom) {
1662         /* this is a *lossless* EOM, so no need to set error, but
1663          * we do need to figure out the actual size */
1664         DEVICE(self)->is_eom = TRUE;
1665     } else {
1666         g_assert_not_reached();
1667         error("not reached");
1668     }
1669
1670     return 0;
1671 }
1672
1673 static int
1674 read_to_connection_impl(
1675     Device *dself,
1676     guint64 size,
1677     guint64 *actual_size,
1678     int     *cancelled,
1679     GMutex  *abort_mutex,
1680     GCond   *abort_cond)
1681 {
1682     NdmpDevice *self = NDMP_DEVICE(dself);
1683     DirectTCPConnectionNDMP *nconn = self->directtcp_conn;
1684     gboolean eom = FALSE, eof = FALSE, eow = FALSE;
1685     ndmp9_mover_state mover_state;
1686     ndmp9_mover_halt_reason mover_halt_reason = NDMP9_MOVER_HALT_NA;
1687     ndmp9_mover_pause_reason mover_pause_reason = NDMP9_MOVER_PAUSE_NA;
1688     guint64 bytes_moved_before, bytes_moved_after;
1689     gchar *err;
1690     int result;
1691
1692     if (actual_size)
1693         *actual_size = 0;
1694
1695     if (device_in_error(self)) return 1;
1696
1697     /* read_to_connection does not support IndirectTCP */
1698     g_assert(self->indirecttcp_sock == -1);
1699
1700     /* if this is false, then the caller did not use use_connection correctly */
1701     g_assert(nconn != NULL);
1702     g_assert(self->ndmp == nconn->ndmp);
1703     g_assert(nconn->mode == NDMP9_MOVER_MODE_WRITE);
1704
1705     if (!ndmp_connection_mover_get_state(self->ndmp,
1706                 &mover_state, &bytes_moved_before, NULL, NULL)) {
1707         set_error_from_ndmp(self);
1708         return 1;
1709     }
1710
1711     /* the mover had best be PAUSED right now */
1712     g_assert(mover_state == NDMP9_MOVER_STATE_PAUSED);
1713
1714     if (!ndmp_connection_mover_set_window(self->ndmp,
1715                 nconn->offset,
1716                 size? size : G_MAXUINT64 - nconn->offset)) {
1717         set_error_from_ndmp(self);
1718         return 1;
1719     }
1720
1721     if (!ndmp_connection_mover_continue(self->ndmp)) {
1722         set_error_from_ndmp(self);
1723         return 1;
1724     }
1725
1726     /* now wait for the mover to pause itself again, or halt on EOF or an error */
1727     result = ndmp_connection_wait_for_notify_with_cond(self->ndmp,
1728                     NULL,
1729                     &mover_halt_reason,
1730                     &mover_pause_reason, NULL,
1731                     cancelled, abort_mutex, abort_cond);
1732     if (result == 1) {
1733         set_error_from_ndmp(self);
1734         return 1;
1735     } else if (result == 2) {
1736         return 2;
1737     }
1738
1739     err = NULL;
1740     if (mover_pause_reason) {
1741         switch (mover_pause_reason) {
1742             case NDMP9_MOVER_PAUSE_EOF:
1743                 eof = TRUE;
1744                 break;
1745
1746             /* ndmjob sends .._SEEK when it should send .._EOW, so deal with
1747                 * both equivalently */
1748             case NDMP9_MOVER_PAUSE_EOW:
1749             case NDMP9_MOVER_PAUSE_SEEK:
1750                 eow = TRUE;
1751                 break;
1752
1753             default:
1754                 err = "got NOTIFY_MOVER_PAUSED, but not because of EOW or SEEK";
1755                 break;
1756         }
1757     } else if (mover_halt_reason) {
1758         switch (mover_halt_reason) {
1759             case NDMP9_MOVER_HALT_CONNECT_CLOSED:
1760                 eof = TRUE;
1761                 break;
1762
1763             default:
1764             case NDMP9_MOVER_HALT_ABORTED:
1765             /* case NDMP9_MOVER_HALT_MEDIA_ERROR: <-- not in ndmjob */
1766             case NDMP9_MOVER_HALT_INTERNAL_ERROR:
1767             case NDMP9_MOVER_HALT_CONNECT_ERROR:
1768                 err = "unexpected NDMP_NOTIFY_MOVER_HALTED";
1769                 break;
1770         }
1771     }
1772
1773     if (err) {
1774         device_set_error(DEVICE(self),
1775             g_strdup_printf("waiting for accept: %s", err),
1776             DEVICE_STATUS_DEVICE_ERROR);
1777         return 1;
1778     }
1779
1780     /* no error, so the mover stopped due to one of EOM (volume out of space),
1781      * EOF (data connection is done), or EOW (maximum part size was written).
1782      * In any case, we want to know how many bytes were written. */
1783
1784     if (!ndmp_connection_mover_get_state(self->ndmp,
1785                 &mover_state, &bytes_moved_after, NULL, NULL)) {
1786         set_error_from_ndmp(self);
1787         return 1;
1788     }
1789     size = bytes_moved_after - bytes_moved_before;
1790     nconn->offset += size;
1791
1792     if (actual_size) {
1793         *actual_size = bytes_moved_after - bytes_moved_before;
1794     }
1795
1796     if (eow) {
1797         ; /* mover finished the whole part -- nothing to report! */
1798     } else if (eof) {
1799         DEVICE(self)->is_eof = TRUE;
1800     } else if (eom) {
1801         /* this is a *lossless* EOM, so no need to set error, but
1802          * we do need to figure out the actual size */
1803         DEVICE(self)->is_eom = TRUE;
1804     } else {
1805         g_assert_not_reached();
1806         error("not reached");
1807     }
1808
1809     return 0;
1810 }
1811
1812 static gboolean
1813 use_connection_impl(
1814     Device *dself,
1815     DirectTCPConnection *conn)
1816 {
1817     NdmpDevice *self = NDMP_DEVICE(dself);
1818     DirectTCPConnectionNDMP *nconn;
1819
1820     /* the device_use_connection_impl wrapper already made sure we're in
1821      * ACCESS_NULL, but we may have opened the tape service already to read
1822      * a label - so close it to be sure */
1823     if (!close_tape_agent(self)) {
1824         /* error was already set by close_tape_agent */
1825         return FALSE;
1826     }
1827
1828     /* we had best not be listening when this is called */
1829     g_assert(!self->listen_addrs);
1830
1831     if (!IS_DIRECTTCP_CONNECTION_NDMP(conn)) {
1832         device_set_error(DEVICE(self),
1833             g_strdup("existing DirectTCPConnection is not compatible with this device"),
1834             DEVICE_STATUS_DEVICE_ERROR);
1835         return FALSE;
1836     }
1837
1838     if (self->directtcp_conn)
1839         g_object_unref(self->directtcp_conn);
1840     self->directtcp_conn = nconn = DIRECTTCP_CONNECTION_NDMP(conn);
1841     g_object_ref(self->directtcp_conn);
1842
1843     /* if this is a different connection, use it */
1844     if (nconn->ndmp != self->ndmp) {
1845         if (self->ndmp)
1846             close_connection(self);
1847         self->ndmp = nconn->ndmp;
1848         g_object_ref(self->ndmp);
1849     }
1850
1851     return TRUE;
1852 }
1853
1854 /*
1855  * Class mechanics
1856  */
1857
1858 static gboolean
1859 ndmp_device_set_username_fn(Device *dself,
1860     DevicePropertyBase *base, GValue *val,
1861     PropertySurety surety, PropertySource source)
1862 {
1863     NdmpDevice *self = NDMP_DEVICE(dself);
1864
1865     amfree(self->ndmp_username);
1866     self->ndmp_username = g_value_dup_string(val);
1867     device_clear_volume_details(dself);
1868
1869     return device_simple_property_set_fn(dself, base, val, surety, source);
1870 }
1871
1872 static gboolean
1873 ndmp_device_set_password_fn(Device *dself,
1874     DevicePropertyBase *base, GValue *val,
1875     PropertySurety surety, PropertySource source)
1876 {
1877     NdmpDevice *self = NDMP_DEVICE(dself);
1878
1879     amfree(self->ndmp_password);
1880     self->ndmp_password = g_value_dup_string(val);
1881     device_clear_volume_details(dself);
1882
1883     return device_simple_property_set_fn(dself, base, val, surety, source);
1884 }
1885
1886 static gboolean
1887 ndmp_device_set_auth_fn(Device *dself,
1888     DevicePropertyBase *base, GValue *val,
1889     PropertySurety surety, PropertySource source)
1890 {
1891     NdmpDevice *self = NDMP_DEVICE(dself);
1892
1893     amfree(self->ndmp_auth);
1894     self->ndmp_auth = g_value_dup_string(val);
1895     device_clear_volume_details(dself);
1896
1897     return device_simple_property_set_fn(dself, base, val, surety, source);
1898 }
1899
1900 static gboolean
1901 ndmp_device_set_verbose_fn(Device *p_self, DevicePropertyBase *base,
1902     GValue *val, PropertySurety surety, PropertySource source)
1903 {
1904     NdmpDevice *self = NDMP_DEVICE(p_self);
1905
1906     self->verbose = g_value_get_boolean(val);
1907
1908     /* if the connection is active, set up verbose logging or turn it off */
1909     if (self->ndmp) {
1910         ndmp_connection_set_verbose(self->ndmp, self->verbose);
1911     }
1912
1913
1914     return device_simple_property_set_fn(p_self, base, val, surety, source);
1915 }
1916
1917 static gboolean
1918 ndmp_device_set_read_block_size_fn(Device *p_self, DevicePropertyBase *base G_GNUC_UNUSED,
1919     GValue *val, PropertySurety surety, PropertySource source)
1920 {
1921     NdmpDevice *self = NDMP_DEVICE(p_self);
1922     gsize read_block_size = g_value_get_uint(val);
1923
1924     if (read_block_size != 0 &&
1925             ((gsize)read_block_size < p_self->block_size ||
1926              (gsize)read_block_size > p_self->max_block_size)) {
1927         device_set_error(p_self,
1928             g_strdup_printf("Error setting READ-BLOCk-SIZE property to '%zu', it must be between %zu and %zu", read_block_size, p_self->block_size, p_self->max_block_size),
1929             DEVICE_STATUS_DEVICE_ERROR);
1930         return FALSE;
1931     }
1932
1933     self->read_block_size = read_block_size;
1934
1935     /* use the READ_BLOCK_SIZE, even if we're invoked to get the old READ_BUFFER_SIZE */
1936     return device_simple_property_set_fn(p_self, base,
1937                                         val, surety, source);
1938 }
1939
1940 static gboolean
1941 ndmp_device_set_indirect_fn(Device *dself,
1942     DevicePropertyBase *base, GValue *val,
1943     PropertySurety surety, PropertySource source)
1944 {
1945     NdmpDevice *self = NDMP_DEVICE(dself);
1946
1947     self->indirect = g_value_get_boolean(val);
1948
1949     return device_simple_property_set_fn(dself, base, val, surety, source);
1950 }
1951
1952 static void
1953 ndmp_device_class_init(NdmpDeviceClass * c G_GNUC_UNUSED)
1954 {
1955     GObjectClass *g_object_class = (GObjectClass*) c;
1956     DeviceClass *device_class = (DeviceClass *)c;
1957
1958     parent_class = g_type_class_ref (TYPE_DEVICE);
1959
1960     device_class->open_device = ndmp_device_open_device;
1961     device_class->read_label = ndmp_device_read_label;
1962     device_class->start = ndmp_device_start;
1963     device_class->finish = ndmp_device_finish;
1964     device_class->eject = ndmp_device_eject;
1965
1966     device_class->start_file = ndmp_device_start_file;
1967     device_class->write_block = ndmp_device_write_block;
1968     device_class->finish_file = ndmp_device_finish_file;
1969
1970     device_class->seek_file = ndmp_device_seek_file;
1971     device_class->seek_block = ndmp_device_seek_block;
1972     device_class->read_block = ndmp_device_read_block;
1973
1974     device_class->directtcp_supported = TRUE;
1975     device_class->listen = listen_impl;
1976     device_class->accept= accept_impl;
1977     device_class->connect= connect_impl;
1978     device_class->write_from_connection = write_from_connection_impl;
1979     device_class->read_to_connection = read_to_connection_impl;
1980     device_class->use_connection = use_connection_impl;
1981
1982     g_object_class->finalize = ndmp_device_finalize;
1983
1984     device_class_register_property(device_class, PROPERTY_NDMP_USERNAME,
1985             PROPERTY_ACCESS_GET_MASK | PROPERTY_ACCESS_SET_BEFORE_START,
1986             device_simple_property_get_fn,
1987             ndmp_device_set_username_fn);
1988
1989     device_class_register_property(device_class, PROPERTY_NDMP_PASSWORD,
1990             PROPERTY_ACCESS_GET_MASK | PROPERTY_ACCESS_SET_BEFORE_START,
1991             device_simple_property_get_fn,
1992             ndmp_device_set_password_fn);
1993
1994     device_class_register_property(device_class, PROPERTY_NDMP_AUTH,
1995             PROPERTY_ACCESS_GET_MASK | PROPERTY_ACCESS_SET_BEFORE_START,
1996             device_simple_property_get_fn,
1997             ndmp_device_set_auth_fn);
1998
1999     device_class_register_property(device_class, PROPERTY_VERBOSE,
2000             PROPERTY_ACCESS_GET_MASK | PROPERTY_ACCESS_SET_MASK,
2001             device_simple_property_get_fn,
2002             ndmp_device_set_verbose_fn);
2003
2004     device_class_register_property(device_class, PROPERTY_INDIRECT,
2005             PROPERTY_ACCESS_GET_MASK | PROPERTY_ACCESS_SET_MASK,
2006             device_simple_property_get_fn,
2007             ndmp_device_set_indirect_fn);
2008
2009     device_class_register_property(device_class, PROPERTY_READ_BLOCK_SIZE,
2010             PROPERTY_ACCESS_GET_MASK | PROPERTY_ACCESS_SET_BEFORE_START,
2011             device_simple_property_get_fn,
2012             ndmp_device_set_read_block_size_fn);
2013 }
2014
2015 static void
2016 ndmp_device_init(NdmpDevice *self)
2017 {
2018     Device *dself = DEVICE(self);
2019     GValue response;
2020
2021     /* begin unconnected */
2022     self->ndmp = NULL;
2023
2024     /* decent defaults */
2025     dself->block_size = 32768;
2026     dself->min_block_size = 32768;
2027     dself->max_block_size = SIZE_MAX;
2028
2029     bzero(&response, sizeof(response));
2030
2031     g_value_init(&response, CONCURRENCY_PARADIGM_TYPE);
2032     g_value_set_enum(&response, CONCURRENCY_PARADIGM_EXCLUSIVE);
2033     device_set_simple_property(dself, PROPERTY_CONCURRENCY,
2034             &response, PROPERTY_SURETY_GOOD, PROPERTY_SOURCE_DETECTED);
2035     g_value_unset(&response);
2036
2037     g_value_init(&response, STREAMING_REQUIREMENT_TYPE);
2038     g_value_set_enum(&response, STREAMING_REQUIREMENT_DESIRED);
2039     device_set_simple_property(dself, PROPERTY_STREAMING,
2040             &response, PROPERTY_SURETY_GOOD, PROPERTY_SOURCE_DETECTED);
2041     g_value_unset(&response);
2042
2043     g_value_init(&response, G_TYPE_BOOLEAN);
2044     g_value_set_boolean(&response, FALSE);
2045     device_set_simple_property(dself, PROPERTY_APPENDABLE,
2046             &response, PROPERTY_SURETY_GOOD, PROPERTY_SOURCE_DETECTED);
2047     g_value_unset(&response);
2048
2049     g_value_init(&response, G_TYPE_BOOLEAN);
2050     g_value_set_boolean(&response, FALSE);
2051     device_set_simple_property(dself, PROPERTY_PARTIAL_DELETION,
2052             &response, PROPERTY_SURETY_GOOD, PROPERTY_SOURCE_DETECTED);
2053     g_value_unset(&response);
2054
2055     g_value_init(&response, G_TYPE_BOOLEAN);
2056     g_value_set_boolean(&response, FALSE);
2057     device_set_simple_property(dself, PROPERTY_FULL_DELETION,
2058             &response, PROPERTY_SURETY_GOOD, PROPERTY_SOURCE_DETECTED);
2059     g_value_unset(&response);
2060
2061     g_value_init(&response, G_TYPE_BOOLEAN);
2062     g_value_set_boolean(&response, TRUE);
2063     device_set_simple_property(dself, PROPERTY_LEOM,
2064             &response, PROPERTY_SURETY_GOOD, PROPERTY_SOURCE_DETECTED);
2065     g_value_unset(&response);
2066
2067     g_value_init(&response, MEDIA_ACCESS_MODE_TYPE);
2068     g_value_set_enum(&response, MEDIA_ACCESS_MODE_READ_WRITE);
2069     device_set_simple_property(dself, PROPERTY_MEDIUM_ACCESS_TYPE,
2070             &response, PROPERTY_SURETY_GOOD, PROPERTY_SOURCE_DETECTED);
2071     g_value_unset(&response);
2072
2073     self->read_block_size = 0;
2074     g_value_init(&response, G_TYPE_UINT);
2075     g_value_set_uint(&response, self->read_block_size);
2076     device_set_simple_property(dself, PROPERTY_READ_BLOCK_SIZE,
2077             &response, PROPERTY_SURETY_GOOD, PROPERTY_SOURCE_DEFAULT);
2078     g_value_unset(&response);
2079
2080     g_value_init(&response, G_TYPE_STRING);
2081     g_value_set_string(&response, "ndmp");
2082     device_set_simple_property(dself, PROPERTY_NDMP_USERNAME,
2083             &response, PROPERTY_SURETY_BAD, PROPERTY_SOURCE_DEFAULT);
2084     g_value_unset(&response);
2085     self->ndmp_username = g_strdup("ndmp");
2086
2087     g_value_init(&response, G_TYPE_STRING);
2088     g_value_set_string(&response, "ndmp");
2089     device_set_simple_property(dself, PROPERTY_NDMP_PASSWORD,
2090             &response, PROPERTY_SURETY_BAD, PROPERTY_SOURCE_DEFAULT);
2091     g_value_unset(&response);
2092     self->ndmp_password = g_strdup("ndmp");
2093
2094     g_value_init(&response, G_TYPE_STRING);
2095     g_value_set_string(&response, "md5");
2096     device_set_simple_property(dself, PROPERTY_NDMP_AUTH,
2097             &response, PROPERTY_SURETY_BAD, PROPERTY_SOURCE_DEFAULT);
2098     g_value_unset(&response);
2099     self->ndmp_auth = g_strdup("md5");
2100
2101     g_value_init(&response, G_TYPE_BOOLEAN);
2102     g_value_set_boolean(&response, FALSE);
2103     device_set_simple_property(dself, PROPERTY_INDIRECT,
2104             &response, PROPERTY_SURETY_GOOD, PROPERTY_SOURCE_DEFAULT);
2105     g_value_unset(&response);
2106     self->indirect = FALSE;
2107
2108     self->indirecttcp_sock = -1;
2109 }
2110
2111 static GType
2112 ndmp_device_get_type(void)
2113 {
2114     static GType type = 0;
2115
2116     if G_UNLIKELY(type == 0) {
2117         static const GTypeInfo info = {
2118             sizeof (NdmpDeviceClass),
2119             (GBaseInitFunc) NULL,
2120             (GBaseFinalizeFunc) NULL,
2121             (GClassInitFunc) ndmp_device_class_init,
2122             (GClassFinalizeFunc) NULL,
2123             NULL /* class_data */,
2124             sizeof (NdmpDevice),
2125             0 /* n_preallocs */,
2126             (GInstanceInitFunc) ndmp_device_init,
2127             NULL
2128         };
2129
2130         type = g_type_register_static (TYPE_DEVICE, "NdmpDevice", &info,
2131                                        (GTypeFlags)0);
2132     }
2133
2134     return type;
2135 }
2136
2137 static Device*
2138 ndmp_device_factory(
2139     char *device_name,
2140     char *device_type,
2141     char *device_node)
2142 {
2143     Device *rval;
2144     g_assert(0 == strcmp(device_type, NDMP_DEVICE_NAME));
2145     rval = DEVICE(g_object_new(TYPE_NDMP_DEVICE, NULL));
2146
2147     device_open_device(rval, device_name, device_type, device_node);
2148     return rval;
2149 }
2150
2151 void
2152 ndmp_device_register(void)
2153 {
2154     static const char * device_prefix_list[] = { NDMP_DEVICE_NAME, NULL };
2155
2156     /* register the device itself */
2157     register_device(ndmp_device_factory, device_prefix_list);
2158
2159     device_property_fill_and_register(&device_property_ndmp_username,
2160                                       G_TYPE_STRING, "ndmp_username",
2161        "Username for access to the NDMP agent");
2162     device_property_fill_and_register(&device_property_ndmp_password,
2163                                       G_TYPE_STRING, "ndmp_password",
2164        "Password for access to the NDMP agent");
2165     device_property_fill_and_register(&device_property_ndmp_auth,
2166                                       G_TYPE_STRING, "ndmp_auth",
2167        "Authentication method for the NDMP agent - md5 (default), text, none, or void");
2168     device_property_fill_and_register(&device_property_indirect,
2169                                       G_TYPE_BOOLEAN, "indirect",
2170        "Use Indirect TCP mode, even if the NDMP server supports "
2171        "window length 0");
2172 }
2173
2174 /*
2175  * DirectTCPConnectionNDMP implementation
2176  */
2177
2178 static char *
2179 directtcp_connection_ndmp_close(DirectTCPConnection *dself)
2180 {
2181     DirectTCPConnectionNDMP *self = DIRECTTCP_CONNECTION_NDMP(dself);
2182     char *rv = NULL;
2183     ndmp9_mover_state state;
2184     guint64 bytes_moved;
2185     ndmp9_mover_halt_reason mover_halt_reason;
2186     gboolean expect_notif = FALSE;
2187
2188     /* based on the current state, we may need to abort or stop the
2189      * mover before closing it */
2190     if (!ndmp_connection_mover_get_state(self->ndmp, &state,
2191                                     &bytes_moved, NULL, NULL)) {
2192         rv = ndmp_connection_err_msg(self->ndmp);
2193         goto error;
2194     }
2195
2196     switch (state) {
2197         case NDMP9_MOVER_STATE_HALTED:
2198             break; /* nothing to do but ndmp_mover_close, below */
2199
2200         case NDMP9_MOVER_STATE_PAUSED:
2201             if (!ndmp_connection_mover_close(self->ndmp)) {
2202                 rv = ndmp_connection_err_msg(self->ndmp);
2203                 goto error;
2204             }
2205             expect_notif = TRUE;
2206             break;
2207
2208         case NDMP9_MOVER_STATE_ACTIVE:
2209         default:
2210             if (!ndmp_connection_mover_abort(self->ndmp)) {
2211                 rv = ndmp_connection_err_msg(self->ndmp);
2212                 goto error;
2213             }
2214             expect_notif = TRUE;
2215             break;
2216     }
2217
2218     /* the spec isn't entirely clear that mover_close and mover_abort should
2219      * generate a NOTIF_MOVER_HALTED, but ndmjob does it */
2220     if (expect_notif) {
2221         if (!ndmp_connection_wait_for_notify(self->ndmp,
2222                 NULL,
2223                 &mover_halt_reason, /* value is ignored.. */
2224                 NULL, NULL)) {
2225             goto error;
2226         }
2227     }
2228
2229     if (!ndmp_connection_mover_stop(self->ndmp)) {
2230         rv = ndmp_connection_err_msg(self->ndmp);
2231         goto error;
2232     }
2233
2234 error:
2235     if (self->ndmp) {
2236         g_object_unref(self->ndmp);
2237         self->ndmp = NULL;
2238     }
2239
2240     return rv;
2241 }
2242
2243 static void
2244 directtcp_connection_ndmp_class_init(DirectTCPConnectionNDMPClass * c)
2245 {
2246     DirectTCPConnectionClass *connc = (DirectTCPConnectionClass *)c;
2247
2248     connc->close = directtcp_connection_ndmp_close;
2249 }
2250
2251 GType
2252 directtcp_connection_ndmp_get_type (void)
2253 {
2254     static GType type = 0;
2255
2256     if G_UNLIKELY(type == 0) {
2257         static const GTypeInfo info = {
2258             sizeof (DirectTCPConnectionNDMPClass),
2259             (GBaseInitFunc) NULL,
2260             (GBaseFinalizeFunc) NULL,
2261             (GClassInitFunc) directtcp_connection_ndmp_class_init,
2262             (GClassFinalizeFunc) NULL,
2263             NULL /* class_data */,
2264             sizeof (DirectTCPConnectionNDMP),
2265             0 /* n_preallocs */,
2266             (GInstanceInitFunc) NULL,
2267             NULL
2268         };
2269
2270         type = g_type_register_static(TYPE_DIRECTTCP_CONNECTION,
2271                                 "DirectTCPConnectionNDMP", &info, (GTypeFlags)0);
2272     }
2273
2274     return type;
2275 }
2276
2277 static DirectTCPConnectionNDMP *
2278 directtcp_connection_ndmp_new(
2279     NDMPConnection *ndmp,
2280     ndmp9_mover_mode mode)
2281 {
2282     DirectTCPConnectionNDMP *dcn = DIRECTTCP_CONNECTION_NDMP(
2283             g_object_new(TYPE_DIRECTTCP_CONNECTION_NDMP, NULL));
2284
2285     /* hang onto a copy of this NDMP connection */
2286     g_object_ref(ndmp);
2287     dcn->ndmp = ndmp;
2288     dcn->mode = mode;
2289     dcn->offset = 0;
2290
2291     return dcn;
2292 }