git.gag.com Git - debian/amanda/blob - server-src/chunker.c

   1 /*
   2  * Amanda, The Advanced Maryland Automatic Network Disk Archiver
   3  * Copyright (c) 1991-1999 University of Maryland at College Park
   4  * All Rights Reserved.
   5  *
   6  * Permission to use, copy, modify, distribute, and sell this software and its
   7  * documentation for any purpose is hereby granted without fee, provided that
   8  * the above copyright notice appear in all copies and that both that
   9  * copyright notice and this permission notice appear in supporting
  10  * documentation, and that the name of U.M. not be used in advertising or
  11  * publicity pertaining to distribution of the software without specific,
  12  * written prior permission.  U.M. makes no representations about the
  13  * suitability of this software for any purpose.  It is provided "as is"
  14  * without express or implied warranty.
  15  *
  16  * U.M. DISCLAIMS ALL WARRANTIES WITH REGARD TO THIS SOFTWARE, INCLUDING ALL
  17  * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS, IN NO EVENT SHALL U.M.
  18  * BE LIABLE FOR ANY SPECIAL, INDIRECT OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
  19  * WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN ACTION
  20  * OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF OR IN
  21  * CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
  22  *
  23  * Authors: the Amanda Development Team.  Its members are listed in a
  24  * file named AUTHORS, in the root directory of this distribution.
  25  */
  26 /* $Id: chunker.c,v 1.36 2006/08/24 11:23:32 martinea Exp $
  27  *
  28  * requests remote amandad processes to dump filesystems
  29  */
  30 #include "amanda.h"
  31 #include "arglist.h"
  32 #include "clock.h"
  33 #include "conffile.h"
  34 #include "event.h"
  35 #include "logfile.h"
  36 #include "packet.h"
  37 #include "protocol.h"
  38 #include "security.h"
  39 #include "stream.h"
  40 #include "fileheader.h"
  41 #include "amfeatures.h"
  42 #include "server_util.h"
  43 #include "util.h"
  44 #include "holding.h"
  45 #include "timestamp.h"
  46 #include "sockaddr-util.h"
  47
  48 #ifndef SEEK_SET
  49 #define SEEK_SET 0
  50 #endif
  51
  52 #ifndef SEEK_CUR
  53 #define SEEK_CUR 1
  54 #endif
  55
  56 #define CONNECT_TIMEOUT 5*60
  57
  58 #define STARTUP_TIMEOUT 60
  59
  60 struct databuf {
  61     int fd;                     /* file to flush to */
  62     char *filename;             /* name of what fd points to */
  63     int filename_seq;           /* for chunking */
  64     off_t split_size;           /* when to chunk */
  65     off_t chunk_size;           /* size of each chunk */
  66     off_t use;                  /* size to use on this disk */
  67     char buf[DISK_BLOCK_BYTES];
  68     char *datain;               /* data buffer markers */
  69     char *dataout;
  70     char *datalimit;
  71 };
  72
  73 static char *handle = NULL;
  74
  75 static char *errstr = NULL;
  76 static int abort_pending;
  77 static off_t dumpsize;
  78 static unsigned long headersize;
  79 static off_t dumpbytes;
  80 static off_t filesize;
  81
  82 static char *hostname = NULL;
  83 static char *diskname = NULL;
  84 static char *qdiskname = NULL;
  85 static char *options = NULL;
  86 static char *progname = NULL;
  87 static int level;
  88 static char *dumpdate = NULL;
  89 static struct cmdargs *command_in_transit = NULL;
  90 static char *chunker_timestamp = NULL;
  91
  92 static dumpfile_t file;
  93
  94 /* local functions */
  95 int main(int, char **);
  96 static ssize_t write_tapeheader(int, dumpfile_t *);
  97 static void databuf_init(struct databuf *, int, char *, off_t, off_t);
  98 static int databuf_flush(struct databuf *);
  99
 100 static int startup_chunker(char *, off_t, off_t, struct databuf *, int *, int *);
 101 static int do_chunk(int, struct databuf *, int, int);
 102
 103 /* we use a function pointer for full_write, so that we can "shim" in
 104  * full_write_with_fake_enospc for testing */
 105 static size_t (*db_full_write)(int fd, const void *buf, size_t count);
 106 static size_t full_write_with_fake_enospc(int fd, const void *buf, size_t count);
 107 static off_t fake_enospc_at_byte = -1;
 108
 109 int
 110 main(
 111     int         argc,
 112     char **     argv)
 113 {
 114     static struct databuf db;
 115     struct cmdargs *cmdargs;
 116     int header_fd;
 117     char *q = NULL;
 118     char *filename = NULL;
 119     off_t chunksize, use;
 120     times_t runtime;
 121     am_feature_t *their_features = NULL;
 122     int a;
 123     config_overrides_t *cfg_ovr = NULL;
 124     char *cfg_opt = NULL;
 125     char *m;
 126     int header_socket;
 127     int data_socket;
 128
 129     /*
 130      * Configure program for internationalization:
 131      *   1) Only set the message locale for now.
 132      *   2) Set textdomain for all amanda related programs to "amanda"
 133      *      We don't want to be forced to support dozens of message catalogs.
 134      */
 135     setlocale(LC_MESSAGES, "C");
 136     textdomain("amanda");
 137
 138     safe_fd(-1, 0);
 139
 140     set_pname("chunker");
 141
 142     dbopen(DBG_SUBDIR_SERVER);
 143
 144     /* Don't die when child closes pipe */
 145     signal(SIGPIPE, SIG_IGN);
 146
 147     add_amanda_log_handler(amanda_log_stderr);
 148     add_amanda_log_handler(amanda_log_trace_log);
 149
 150     cfg_ovr = extract_commandline_config_overrides(&argc, &argv);
 151
 152     if (argc > 1)
 153         cfg_opt = argv[1];
 154
 155     set_config_overrides(cfg_ovr);
 156     config_init(CONFIG_INIT_EXPLICIT_NAME | CONFIG_INIT_USE_CWD, cfg_opt);
 157
 158     if (config_errors(NULL) >= CFGERR_WARNINGS) {
 159         config_print_errors();
 160         if (config_errors(NULL) >= CFGERR_ERRORS) {
 161             g_critical(_("errors processing config file"));
 162         }
 163     }
 164
 165     safe_cd(); /* do this *after* config_init() */
 166
 167     check_running_as(RUNNING_AS_DUMPUSER);
 168
 169     dbrename(get_config_name(), DBG_SUBDIR_SERVER);
 170
 171     log_add(L_INFO, "%s pid %ld", get_pname(), (long)getpid());
 172     g_fprintf(stderr,
 173             _("%s: pid %ld executable %s version %s\n"),
 174             get_pname(), (long) getpid(),
 175             argv[0], VERSION);
 176     fflush(stderr);
 177
 178     /* now, make sure we are a valid user */
 179
 180     signal(SIGPIPE, SIG_IGN);
 181     signal(SIGCHLD, SIG_IGN);
 182
 183     cmdargs = getcmd();
 184     if(cmdargs->cmd == START) {
 185         if(cmdargs->argc <= 1)
 186             error(_("error [dumper START: not enough args: timestamp]"));
 187         chunker_timestamp = newstralloc(chunker_timestamp, cmdargs->argv[1]);
 188     }
 189     else {
 190         log_add(L_INFO, "%s pid %ld", get_pname(), (long)getpid());
 191         error(_("Didn't get START command"));
 192     }
 193     free_cmdargs(cmdargs);
 194
 195     /* set up a fake ENOSPC for testing purposes.  Note that this counts
 196      * headers as well as data written to disk. */
 197     if (getenv("CHUNKER_FAKE_ENOSPC_AT")) {
 198         char *env = getenv("CHUNKER_FAKE_ENOSPC_AT");
 199         fake_enospc_at_byte = (off_t)atoi(env); /* these values are never > MAXINT */
 200         db_full_write = full_write_with_fake_enospc;
 201         g_debug("will trigger fake ENOSPC at byte %d", (int)fake_enospc_at_byte);
 202     } else {
 203         db_full_write = full_write;
 204     }
 205
 206 /*    do {*/
 207         cmdargs = getcmd();
 208
 209         switch(cmdargs->cmd) {
 210         case QUIT:
 211             break;
 212
 213         case PORT_WRITE:
 214             /*
 215              * PORT-WRITE
 216              *   handle
 217              *   filename
 218              *   host
 219              *   features
 220              *   disk
 221              *   level
 222              *   dumpdate
 223              *   chunksize
 224              *   progname
 225              *   use
 226              *   options
 227              */
 228             a = 1;
 229
 230             if(a >= cmdargs->argc) {
 231                 error(_("error [chunker PORT-WRITE: not enough args: handle]"));
 232                 /*NOTREACHED*/
 233             }
 234             handle = newstralloc(handle, cmdargs->argv[a++]);
 235
 236             if(a >= cmdargs->argc) {
 237                 error(_("error [chunker PORT-WRITE: not enough args: filename]"));
 238                 /*NOTREACHED*/
 239             }
 240             filename = newstralloc(filename, cmdargs->argv[a++]);
 241
 242             if(a >= cmdargs->argc) {
 243                 error(_("error [chunker PORT-WRITE: not enough args: hostname]"));
 244                 /*NOTREACHED*/
 245             }
 246             hostname = newstralloc(hostname, cmdargs->argv[a++]);
 247
 248             if(a >= cmdargs->argc) {
 249                 error(_("error [chunker PORT-WRITE: not enough args: features]"));
 250                 /*NOTREACHED*/
 251             }
 252             am_release_feature_set(their_features);
 253             their_features = am_string_to_feature(cmdargs->argv[a++]);
 254             if (!their_features) {
 255                 error(_("error [chunker PORT-WRITE: invalid feature string]"));
 256                 /*NOTREACHED*/
 257             }
 258
 259             if(a >= cmdargs->argc) {
 260                 error(_("error [chunker PORT-WRITE: not enough args: diskname]"));
 261                 /*NOTREACHED*/
 262             }
 263             diskname = newstralloc(diskname, cmdargs->argv[a++]);
 264             if (qdiskname)
 265                 amfree(qdiskname);
 266             qdiskname = quote_string(diskname); /* qdiskname is a global */
 267
 268             if(a >= cmdargs->argc) {
 269                 error(_("error [chunker PORT-WRITE: not enough args: level]"));
 270                 /*NOTREACHED*/
 271             }
 272             level = atoi(cmdargs->argv[a++]);
 273
 274             if(a >= cmdargs->argc) {
 275                 error(_("error [chunker PORT-WRITE: not enough args: dumpdate]"));
 276                 /*NOTREACHED*/
 277             }
 278             dumpdate = newstralloc(dumpdate, cmdargs->argv[a++]);
 279
 280             if(a >= cmdargs->argc) {
 281                 error(_("error [chunker PORT-WRITE: not enough args: chunksize]"));
 282                 /*NOTREACHED*/
 283             }
 284             chunksize = OFF_T_ATOI(cmdargs->argv[a++]);
 285             chunksize = am_floor(chunksize, (off_t)DISK_BLOCK_KB);
 286
 287             if(a >= cmdargs->argc) {
 288                 error(_("error [chunker PORT-WRITE: not enough args: progname]"));
 289                 /*NOTREACHED*/
 290             }
 291             progname = newstralloc(progname, cmdargs->argv[a++]);
 292
 293             if(a >= cmdargs->argc) {
 294                 error(_("error [chunker PORT-WRITE: not enough args: use]"));
 295                 /*NOTREACHED*/
 296             }
 297             use = am_floor(OFF_T_ATOI(cmdargs->argv[a++]), DISK_BLOCK_KB);
 298
 299             if(a >= cmdargs->argc) {
 300                 error(_("error [chunker PORT-WRITE: not enough args: options]"));
 301                 /*NOTREACHED*/
 302             }
 303             options = newstralloc(options, cmdargs->argv[a++]);
 304
 305             if(a != cmdargs->argc) {
 306                 error(_("error [chunker PORT-WRITE: too many args: %d != %d]"),
 307                       cmdargs->argc, a);
 308                 /*NOTREACHED*/
 309             }
 310
 311             if ((header_fd = startup_chunker(filename, use, chunksize, &db,
 312                                              &header_socket, &data_socket)) < 0) {
 313                 q = quote_string(vstrallocf(_("[chunker startup failed: %s]"), errstr));
 314                 putresult(TRYAGAIN, "%s %s\n", handle, q);
 315                 error("startup_chunker failed: %s", errstr);
 316             }
 317             command_in_transit = NULL;
 318             if (header_fd >= 0 && do_chunk(header_fd, &db, header_socket, data_socket)) {
 319                 char kb_str[NUM_STR_SIZE];
 320                 char kps_str[NUM_STR_SIZE];
 321                 double rt;
 322
 323                 runtime = stopclock();
 324                 rt = g_timeval_to_double(runtime);
 325                 g_snprintf(kb_str, SIZEOF(kb_str), "%lld",
 326                          (long long)(dumpsize - (off_t)headersize));
 327                 g_snprintf(kps_str, SIZEOF(kps_str), "%3.1lf",
 328                                 isnormal(rt) ? (double)dumpsize / rt : 0.0);
 329                 errstr = newvstrallocf(errstr, "sec %s kb %s kps %s",
 330                                 walltime_str(runtime), kb_str, kps_str);
 331                 m = vstrallocf("[%s]", errstr);
 332                 q = quote_string(m);
 333                 amfree(m);
 334                 free_cmdargs(cmdargs);
 335                 if(command_in_transit != NULL) {
 336                     cmdargs = command_in_transit;
 337                     command_in_transit = NULL;
 338                 } else {
 339                     cmdargs = getcmd();
 340                 }
 341                 switch(cmdargs->cmd) {
 342                 case DONE:
 343                     putresult(DONE, "%s %lld %s\n", handle,
 344                              (long long)(dumpsize - (off_t)headersize), q);
 345                     log_add(L_SUCCESS, "%s %s %s %d [%s]",
 346                             hostname, qdiskname, chunker_timestamp, level, errstr);
 347                     break;
 348                 case BOGUS:
 349                 case TRYAGAIN:
 350                 case FAILED:
 351                 case ABORT_FINISHED:
 352                     if(dumpsize > (off_t)DISK_BLOCK_KB) {
 353                         putresult(PARTIAL, "%s %lld %s\n", handle,
 354                                  (long long)(dumpsize - (off_t)headersize),
 355                                  q);
 356                         log_add(L_PARTIAL, "%s %s %s %d [%s]",
 357                                 hostname, qdiskname, chunker_timestamp, level, errstr);
 358                     }
 359                     else {
 360                         errstr = newvstrallocf(errstr,
 361                                         _("dumper returned %s"), cmdstr[cmdargs->cmd]);
 362                         amfree(q);
 363                         m = vstrallocf("[%s]",errstr);
 364                         q = quote_string(m);
 365                         amfree(m);
 366                         putresult(FAILED, "%s %s\n", handle, q);
 367                         log_add(L_FAIL, "%s %s %s %d [%s]",
 368                                 hostname, qdiskname, chunker_timestamp, level, errstr);
 369                     }
 370                 default: break;
 371                 }
 372                 amfree(q);
 373             } else if (header_fd != -2) {
 374                 if(q == NULL) {
 375                     m = vstrallocf("[%s]", errstr);
 376                     q = quote_string(m);
 377                     amfree(m);
 378                 }
 379                 if(!abort_pending) {
 380                     putresult(FAILED, "%s %s\n", handle, q);
 381                 }
 382                 log_add(L_FAIL, "%s %s %s %d [%s]",
 383                         hostname, qdiskname, chunker_timestamp, level, errstr);
 384                 amfree(q);
 385             }
 386             amfree(filename);
 387             amfree(db.filename);
 388             break;
 389
 390         default:
 391             if(cmdargs->argc >= 1) {
 392                 q = quote_string(cmdargs->argv[0]);
 393             } else {
 394                 q = stralloc(_("(no input?)"));
 395             }
 396             putresult(BAD_COMMAND, "%s\n", q);
 397             amfree(q);
 398             break;
 399         }
 400
 401 /*    } while(cmdargs->cmd != QUIT); */
 402
 403     log_add(L_INFO, "pid-done %ld", (long)getpid());
 404
 405     amfree(errstr);
 406     amfree(chunker_timestamp);
 407     amfree(handle);
 408     amfree(hostname);
 409     amfree(diskname);
 410     amfree(qdiskname);
 411     amfree(dumpdate);
 412     amfree(progname);
 413     amfree(options);
 414     free_cmdargs(cmdargs);
 415     if (command_in_transit)
 416         free_cmdargs(command_in_transit);
 417     am_release_feature_set(their_features);
 418     their_features = NULL;
 419
 420     dbclose();
 421
 422     return (0); /* exit */
 423 }
 424
 425 /*
 426  * Returns a file descriptor to the incoming port
 427  * on success, or -1 on error.
 428  */
 429 static int
 430 startup_chunker(
 431     char *              filename,
 432     off_t               use,
 433     off_t               chunksize,
 434     struct databuf *    db,
 435     int                *headersocket,
 436     int                *datasocket)
 437 {
 438     int header_fd, outfd;
 439     char *tmp_filename, *pc;
 440     in_port_t header_port, data_port;
 441     int header_socket, data_socket;
 442     int result;
 443     struct addrinfo *res;
 444     struct addrinfo *res_addr;
 445     sockaddr_union  *addr = NULL;
 446     sockaddr_union   data_addr;
 447
 448     header_port = 0;
 449     data_port = 0;
 450     if ((result = resolve_hostname("localhost", 0, &res, NULL) != 0)) {
 451         errstr = newvstrallocf(errstr, _("could not resolve localhost: %s"),
 452                                gai_strerror(result));
 453         return -1;
 454     }
 455     for (res_addr = res; res_addr != NULL; res_addr = res_addr->ai_next) {
 456         g_debug("ra: %s\n", str_sockaddr((sockaddr_union*)res_addr->ai_addr));
 457         if (res_addr->ai_family == AF_INET) {
 458             addr = (sockaddr_union *)res_addr->ai_addr;
 459             break;
 460         }
 461     }
 462     if (!addr) {
 463         addr = (sockaddr_union *)res->ai_addr;
 464         g_debug("addr: %s\n", str_sockaddr(addr));
 465     }
 466
 467     header_socket = stream_server(SU_GET_FAMILY(addr), &header_port, 0,
 468                                 STREAM_BUFSIZE, 0);
 469     data_socket = stream_server(SU_GET_FAMILY(addr), &data_port, 0,
 470                                 STREAM_BUFSIZE, 0);
 471     copy_sockaddr(&data_addr, addr);
 472
 473     SU_SET_PORT(&data_addr, data_port);
 474
 475     if (res) freeaddrinfo(res);
 476
 477     if (header_socket < 0) {
 478         errstr = vstrallocf(_("error creating header stream server: %s"), strerror(errno));
 479         aclose(data_socket);
 480         return -1;
 481     }
 482
 483     if (data_socket < 0) {
 484         errstr = vstrallocf(_("error creating data stream server: %s"), strerror(errno));
 485         aclose(header_socket);
 486         return -1;
 487     }
 488
 489     putresult(PORT, "%d %s\n", header_port, str_sockaddr(&data_addr));
 490
 491     header_fd = stream_accept(header_socket, CONNECT_TIMEOUT, 0,
 492                               STREAM_BUFSIZE);
 493     if (header_fd == -1) {
 494         errstr = vstrallocf(_("error accepting header stream: %s"),
 495                             strerror(errno));
 496         aclose(header_socket);
 497         aclose(data_socket);
 498         return -1;
 499     }
 500
 501     tmp_filename = vstralloc(filename, ".tmp", NULL);
 502     pc = strrchr(tmp_filename, '/');
 503     g_assert(pc != NULL);
 504     *pc = '\0';
 505     mkholdingdir(tmp_filename);
 506     *pc = '/';
 507     if ((outfd = open(tmp_filename, O_RDWR|O_CREAT|O_TRUNC, 0600)) < 0) {
 508         int save_errno = errno;
 509         char *m = vstrallocf(_("holding file \"%s\": %s"),
 510                          tmp_filename,
 511                          strerror(errno));
 512
 513         errstr = quote_string(m);
 514         amfree(m);
 515         amfree(tmp_filename);
 516         aclose(header_fd);
 517         aclose(data_socket);
 518         if(save_errno == ENOSPC) {
 519             putresult(NO_ROOM, "%s %lld\n",
 520                       handle, (long long)use);
 521             return -2;
 522         } else {
 523             return -1;
 524         }
 525     }
 526     amfree(tmp_filename);
 527     databuf_init(db, outfd, filename, use, chunksize);
 528     db->filename_seq++;
 529     *headersocket = header_socket;
 530     *datasocket = data_socket;
 531     return header_fd;
 532 }
 533
 534 static int
 535 do_chunk(
 536     int                 header_fd,
 537     struct databuf *    db,
 538     int                 header_socket,
 539     int                 data_socket)
 540 {
 541     size_t nread;
 542     int    data_fd;
 543     char header_buf[DISK_BLOCK_BYTES];
 544
 545     startclock();
 546
 547     dumpsize = dumpbytes = filesize = (off_t)0;
 548     headersize = 0;
 549     memset(header_buf, 0, sizeof(header_buf));
 550
 551     /*
 552      * The first thing we should receive is the file header, which we
 553      * need to save into "file", as well as write out.  Later, the
 554      * chunk code will rewrite it.
 555      */
 556     nread = full_read(header_fd, header_buf, SIZEOF(header_buf));
 557     aclose(header_fd);
 558     aclose(header_socket);
 559     if (nread != sizeof(header_buf)) {
 560         if(errno != 0) {
 561             errstr = vstrallocf(_("cannot read header: %s"), strerror(errno));
 562         } else {
 563             errstr = vstrallocf(_("cannot read header: got %zd bytes instead of %zd"),
 564                                 nread, sizeof(header_buf));
 565         }
 566         aclose(data_socket);
 567         return 0;
 568     }
 569     parse_file_header(header_buf, &file, (size_t)nread);
 570     if(write_tapeheader(db->fd, &file)) {
 571         int save_errno = errno;
 572         char *m = vstrallocf(_("write_tapeheader file %s: %s"),
 573                          db->filename, strerror(errno));
 574         errstr = quote_string(m);
 575         amfree(m);
 576         if(save_errno == ENOSPC) {
 577             putresult(NO_ROOM, "%s %lld\n", handle,
 578                       (long long)(db->use+db->split_size-dumpsize));
 579         }
 580         aclose(data_socket);
 581         return 0;
 582     }
 583     dumpsize += (off_t)DISK_BLOCK_KB;
 584     filesize = (off_t)DISK_BLOCK_KB;
 585     headersize += DISK_BLOCK_KB;
 586
 587     /* open the data socket */
 588     data_fd = stream_accept(data_socket, CONNECT_TIMEOUT, 0, STREAM_BUFSIZE);
 589
 590     if (data_fd == -1) {
 591         errstr = vstrallocf(_("error accepting data stream: %s"),
 592                             strerror(errno));
 593         aclose(data_socket);
 594         return 0;
 595     }
 596
 597     /*
 598      * We've written the file header.  Now, just write data until the
 599      * end.
 600      */
 601     while ((nread = full_read(data_fd, db->buf,
 602                              (size_t)(db->datalimit - db->datain))) > 0) {
 603         db->datain += nread;
 604         while(db->dataout < db->datain) {
 605             if(!databuf_flush(db)) {
 606                 aclose(data_fd);
 607                 aclose(data_socket);
 608                 return 0;
 609             }
 610         }
 611     }
 612     while(db->dataout < db->datain) {
 613         if(!databuf_flush(db)) {
 614             aclose(data_fd);
 615             aclose(data_socket);
 616             return 0;
 617         }
 618     }
 619     if(dumpbytes > (off_t)0) {
 620         dumpsize += (off_t)1;                   /* count partial final KByte */
 621         filesize += (off_t)1;
 622     }
 623     aclose(data_fd);
 624     aclose(data_socket);
 625     return 1;
 626 }
 627
 628 /*
 629  * Initialize a databuf.  Takes a writeable file descriptor.
 630  */
 631 static void
 632 databuf_init(
 633     struct databuf *    db,
 634     int                 fd,
 635     char *              filename,
 636     off_t               use,
 637     off_t               chunk_size)
 638 {
 639     db->fd = fd;
 640     db->filename = stralloc(filename);
 641     db->filename_seq = (off_t)0;
 642     db->chunk_size = chunk_size;
 643     db->split_size = (db->chunk_size > use) ? use : db->chunk_size;
 644     db->use = (use > db->split_size) ? use - db->split_size : (off_t)0;
 645     db->datain = db->dataout = db->buf;
 646     db->datalimit = db->buf + SIZEOF(db->buf);
 647 }
 648
 649
 650 /*
 651  * Write out the buffer to the backing file
 652  */
 653 static int
 654 databuf_flush(
 655     struct databuf *    db)
 656 {
 657     struct cmdargs *cmdargs = NULL;
 658     int rc = 1;
 659     size_t size_to_write;
 660     size_t written;
 661     off_t left_in_chunk;
 662     char *arg_filename = NULL;
 663     char *new_filename = NULL;
 664     char *tmp_filename = NULL;
 665     char sequence[NUM_STR_SIZE];
 666     int newfd;
 667     filetype_t save_type;
 668     char *q;
 669     int a;
 670     char *pc;
 671
 672     /*
 673      * If there's no data, do nothing.
 674      */
 675     if (db->dataout >= db->datain) {
 676         goto common_exit;
 677     }
 678
 679     /*
 680      * See if we need to split this file.
 681      */
 682     while (db->split_size > (off_t)0 && dumpsize >= db->split_size) {
 683         if( db->use == (off_t)0 ) {
 684             /*
 685              * Probably no more space on this disk.  Request some more.
 686              */
 687             putresult(RQ_MORE_DISK, "%s\n", handle);
 688             cmdargs = getcmd();
 689             if(command_in_transit == NULL &&
 690                (cmdargs->cmd == DONE || cmdargs->cmd == TRYAGAIN || cmdargs->cmd == FAILED)) {
 691                 command_in_transit = cmdargs;
 692                 cmdargs = getcmd();
 693             }
 694             if(cmdargs->cmd == CONTINUE) {
 695                 /*
 696                  * CONTINUE
 697                  *   serial
 698                  *   filename
 699                  *   chunksize
 700                  *   use
 701                  */
 702                 a = 2; /* skip CONTINUE and serial */
 703
 704                 if(a >= cmdargs->argc) {
 705                     error(_("error [chunker CONTINUE: not enough args: filename]"));
 706                     /*NOTREACHED*/
 707                 }
 708                 arg_filename = newstralloc(arg_filename, cmdargs->argv[a++]);
 709
 710                 if(a >= cmdargs->argc) {
 711                     error(_("error [chunker CONTINUE: not enough args: chunksize]"));
 712                     /*NOTREACHED*/
 713                 }
 714                 db->chunk_size = OFF_T_ATOI(cmdargs->argv[a++]);
 715                 db->chunk_size = am_floor(db->chunk_size, (off_t)DISK_BLOCK_KB);
 716
 717                 if(a >= cmdargs->argc) {
 718                     error(_("error [chunker CONTINUE: not enough args: use]"));
 719                     /*NOTREACHED*/
 720                 }
 721                 db->use = OFF_T_ATOI(cmdargs->argv[a++]);
 722
 723                 if(a != cmdargs->argc) {
 724                     error(_("error [chunker CONTINUE: too many args: %d != %d]"),
 725                           cmdargs->argc, a);
 726                     /*NOTREACHED*/
 727                 }
 728
 729                 if(strcmp(db->filename, arg_filename) == 0) {
 730                     /*
 731                      * Same disk, so use what room is left up to the
 732                      * next chunk boundary or the amount we were given,
 733                      * whichever is less.
 734                      */
 735                     left_in_chunk = db->chunk_size - filesize;
 736                     if(left_in_chunk > db->use) {
 737                         db->split_size += db->use;
 738                         db->use = (off_t)0;
 739                     } else {
 740                         db->split_size += left_in_chunk;
 741                         db->use -= left_in_chunk;
 742                     }
 743                     if(left_in_chunk > (off_t)0) {
 744                         /*
 745                          * We still have space in this chunk.
 746                          */
 747                         break;
 748                     }
 749                 } else {
 750                     /*
 751                      * Different disk, so use new file.
 752                      */
 753                     db->filename = newstralloc(db->filename, arg_filename);
 754                 }
 755             } else if(cmdargs->cmd == ABORT) {
 756                 abort_pending = 1;
 757                 errstr = newstralloc(errstr, cmdargs->argv[1]);
 758                 putresult(ABORT_FINISHED, "%s\n", handle);
 759                 rc = 0;
 760                 goto common_exit;
 761             } else {
 762                 if(cmdargs->argc >= 1) {
 763                     q = quote_string(cmdargs->argv[0]);
 764                 } else {
 765                     q = stralloc(_("(no input?)"));
 766                 }
 767                 error(_("error [bad command after RQ-MORE-DISK: \"%s\"]"), q);
 768                 /*NOTREACHED*/
 769             }
 770         }
 771
 772         /*
 773          * Time to use another file.
 774          */
 775
 776         /*
 777          * First, open the new chunk file, and give it a new header
 778          * that has no cont_filename pointer.
 779          */
 780         g_snprintf(sequence, SIZEOF(sequence), "%d", db->filename_seq);
 781         new_filename = newvstralloc(new_filename,
 782                                     db->filename,
 783                                     ".",
 784                                     sequence,
 785                                     NULL);
 786         tmp_filename = newvstralloc(tmp_filename,
 787                                     new_filename,
 788                                     ".tmp",
 789                                     NULL);
 790         pc = strrchr(tmp_filename, '/');
 791         g_assert(pc != NULL); /* Only a problem if db->filename has no /. */
 792         *pc = '\0';
 793         mkholdingdir(tmp_filename);
 794         *pc = '/';
 795         newfd = open(tmp_filename, O_RDWR|O_CREAT|O_TRUNC, 0600);
 796         if (newfd == -1) {
 797             int save_errno = errno;
 798             char *m;
 799
 800             if(save_errno == ENOSPC) {
 801                 putresult(NO_ROOM, "%s %lld\n", handle,
 802                           (long long)(db->use+db->split_size-dumpsize));
 803                 db->use = (off_t)0;                     /* force RQ_MORE_DISK */
 804                 db->split_size = dumpsize;
 805                 continue;
 806             }
 807             m = vstrallocf(_("creating chunk holding file \"%s\": %s"),
 808                              tmp_filename,
 809                              strerror(errno));
 810             errstr = quote_string(m);
 811             amfree(m);
 812             aclose(db->fd);
 813             rc = 0;
 814             goto common_exit;
 815         }
 816         save_type = file.type;
 817         file.type = F_CONT_DUMPFILE;
 818         file.cont_filename[0] = '\0';
 819         if(write_tapeheader(newfd, &file)) {
 820             int save_errno = errno;
 821             char *m;
 822
 823             aclose(newfd);
 824             if(save_errno == ENOSPC) {
 825                 if (unlink(tmp_filename) < 0) {
 826                     g_debug("could not delete '%s'; ignoring", tmp_filename);
 827                 }
 828                 putresult(NO_ROOM, "%s %lld\n", handle,
 829                           (long long)(db->use+db->split_size-dumpsize));
 830                 db->use = (off_t)0;                     /* force RQ_MORE DISK */
 831                 db->split_size = dumpsize;
 832                 file.type = save_type;
 833                 continue;
 834             }
 835             m = vstrallocf(_("write_tapeheader file %s: %s"),
 836                              tmp_filename,
 837                              strerror(errno));
 838             errstr = quote_string(m);
 839             amfree(m);
 840             rc = 0;
 841             goto common_exit;
 842         }
 843
 844         /*
 845          * Now, update the header of the current file to point
 846          * to the next chunk, and then close it.
 847          */
 848         if (lseek(db->fd, (off_t)0, SEEK_SET) < (off_t)0) {
 849             char *m = vstrallocf(_("lseek holding file %s: %s"),
 850                              db->filename,
 851                              strerror(errno));
 852             errstr = quote_string(m);
 853             amfree(m);
 854             aclose(newfd);
 855             rc = 0;
 856             goto common_exit;
 857         }
 858
 859         file.type = save_type;
 860         strncpy(file.cont_filename, new_filename, SIZEOF(file.cont_filename));
 861         file.cont_filename[SIZEOF(file.cont_filename)-1] = '\0';
 862         if(write_tapeheader(db->fd, &file)) {
 863             char * m = vstrallocf(_("write_tapeheader file \"%s\": %s"),
 864                              db->filename,
 865                              strerror(errno));
 866             errstr = quote_string(m);
 867             amfree(m);
 868             aclose(newfd);
 869             unlink(tmp_filename);
 870             rc = 0;
 871             goto common_exit;
 872         }
 873         file.type = F_CONT_DUMPFILE;
 874
 875         /*
 876          * Now shift the file descriptor.
 877          */
 878         aclose(db->fd);
 879         db->fd = newfd;
 880         newfd = -1;
 881
 882         /*
 883          * Update when we need to chunk again
 884          */
 885         if(db->use <= (off_t)DISK_BLOCK_KB) {
 886             /*
 887              * Cheat and use one more block than allowed so we can make
 888              * some progress.
 889              */
 890             db->split_size += (off_t)(2 * DISK_BLOCK_KB);
 891             db->use = (off_t)0;
 892         } else if(db->chunk_size > db->use) {
 893             db->split_size += db->use;
 894             db->use = (off_t)0;
 895         } else {
 896             db->split_size += db->chunk_size;
 897             db->use -= db->chunk_size;
 898         }
 899
 900
 901         amfree(tmp_filename);
 902         amfree(new_filename);
 903         dumpsize += (off_t)DISK_BLOCK_KB;
 904         filesize = (off_t)DISK_BLOCK_KB;
 905         headersize += DISK_BLOCK_KB;
 906         db->filename_seq++;
 907     }
 908
 909     /*
 910      * Write out the buffer
 911      */
 912     size_to_write = (size_t)(db->datain - db->dataout);
 913     written = db_full_write(db->fd, db->dataout, size_to_write);
 914     if (written > 0) {
 915         db->dataout += written;
 916         dumpbytes += (off_t)written;
 917     }
 918     dumpsize += (dumpbytes / (off_t)1024);
 919     filesize += (dumpbytes / (off_t)1024);
 920     dumpbytes %= 1024;
 921     if (written < size_to_write) {
 922         if (errno != ENOSPC) {
 923             char *m = vstrallocf(_("data write: %s"), strerror(errno));
 924             errstr = quote_string(m);
 925             amfree(m);
 926             rc = 0;
 927             goto common_exit;
 928         }
 929
 930         /*
 931          * NO-ROOM is informational only.  Later, RQ_MORE_DISK will be
 932          * issued to use another holding disk.
 933          */
 934         putresult(NO_ROOM, "%s %lld\n", handle,
 935                   (long long)(db->use+db->split_size-dumpsize));
 936         db->use = (off_t)0;                             /* force RQ_MORE_DISK */
 937         db->split_size = dumpsize;
 938         goto common_exit;
 939     }
 940     if (db->datain == db->dataout) {
 941         /*
 942          * We flushed the whole buffer so reset to use it all.
 943          */
 944         db->datain = db->dataout = db->buf;
 945     }
 946
 947 common_exit:
 948
 949     if (cmdargs)
 950         free_cmdargs(cmdargs);
 951     amfree(new_filename);
 952     /*@i@*/ amfree(tmp_filename);
 953     amfree(arg_filename);
 954     return rc;
 955 }
 956
 957
 958 /*
 959  * Send an Amanda dump header to the output file and set file->blocksize
 960  */
 961 static ssize_t
 962 write_tapeheader(
 963     int         outfd,
 964     dumpfile_t *file)
 965 {
 966     char *buffer;
 967     size_t written;
 968
 969     file->blocksize = DISK_BLOCK_BYTES;
 970     if (debug_chunker > 1)
 971         dump_dumpfile_t(file);
 972     buffer = build_header(file, NULL, DISK_BLOCK_BYTES);
 973     if (!buffer) /* this shouldn't happen */
 974         error(_("header does not fit in %zd bytes"), (size_t)DISK_BLOCK_BYTES);
 975
 976     written = db_full_write(outfd, buffer, DISK_BLOCK_BYTES);
 977     amfree(buffer);
 978     if(written == DISK_BLOCK_BYTES) return 0;
 979
 980     /* fake ENOSPC when we get a short write without errno set */
 981     if(errno == 0)
 982         errno = ENOSPC;
 983
 984     return (ssize_t)-1;
 985 }
 986
 987 static size_t
 988 full_write_with_fake_enospc(
 989     int fd,
 990     const void *buf,
 991     size_t count)
 992 {
 993     size_t rc;
 994
 995     //g_debug("HERE %zd %zd", count, (size_t)fake_enospc_at_byte);
 996
 997     if (count <= (size_t)fake_enospc_at_byte) {
 998         fake_enospc_at_byte -= count;
 999         return full_write(fd, buf, count);
1000     }
1001
1002     /* if we get here, the caller has requested a size that is less
1003      * than fake_enospc_at_byte. */
1004     count = fake_enospc_at_byte;
1005     g_debug("returning fake ENOSPC");
1006
1007     if (fake_enospc_at_byte) {
1008         rc = full_write(fd, buf, fake_enospc_at_byte);
1009         if (rc == (size_t)fake_enospc_at_byte) {
1010             /* full_write succeeded, so fake a failure */
1011             errno = ENOSPC;
1012         }
1013     } else {
1014         /* no bytes to write; just fake an error */
1015         errno = ENOSPC;
1016         rc = 0;
1017     }
1018
1019     /* switch back to calling full_write directly */
1020     fake_enospc_at_byte = -1;
1021     db_full_write = full_write;
1022     return rc;
1023 }