git.sur5r.net Git - bacula/bacula/blob - bacula/src/stored/spool.c

   1 /*
   2    Bacula® - The Network Backup Solution
   3
   4    Copyright (C) 2004-2008 Free Software Foundation Europe e.V.
   5
   6    The main author of Bacula is Kern Sibbald, with contributions from
   7    many others, a complete list can be found in the file AUTHORS.
   8    This program is Free Software; you can redistribute it and/or
   9    modify it under the terms of version two of the GNU General Public
  10    License as published by the Free Software Foundation and included
  11    in the file LICENSE.
  12
  13    This program is distributed in the hope that it will be useful, but
  14    WITHOUT ANY WARRANTY; without even the implied warranty of
  15    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
  16    General Public License for more details.
  17
  18    You should have received a copy of the GNU General Public License
  19    along with this program; if not, write to the Free Software
  20    Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA
  21    02110-1301, USA.
  22
  23    Bacula® is a registered trademark of Kern Sibbald.
  24    The licensor of Bacula is the Free Software Foundation Europe
  25    (FSFE), Fiduciary Program, Sumatrastrasse 25, 8006 Zürich,
  26    Switzerland, email:ftf@fsfeurope.org.
  27 */
  28 /*
  29  *  Spooling code
  30  *
  31  *      Kern Sibbald, March 2004
  32  *
  33  *  Version $Id$
  34  */
  35
  36 #include "bacula.h"
  37 #include "stored.h"
  38
  39 /* Forward referenced subroutines */
  40 static void make_unique_data_spool_filename(DCR *dcr, POOLMEM **name);
  41 static bool open_data_spool_file(DCR *dcr);
  42 static bool close_data_spool_file(DCR *dcr);
  43 static bool despool_data(DCR *dcr, bool commit);
  44 static int  read_block_from_spool_file(DCR *dcr);
  45 static bool open_attr_spool_file(JCR *jcr, BSOCK *bs);
  46 static bool close_attr_spool_file(JCR *jcr, BSOCK *bs);
  47 static bool write_spool_header(DCR *dcr);
  48 static bool write_spool_data(DCR *dcr);
  49
  50 struct spool_stats_t {
  51    uint32_t data_jobs;                /* current jobs spooling data */
  52    uint32_t attr_jobs;
  53    uint32_t total_data_jobs;          /* total jobs to have spooled data */
  54    uint32_t total_attr_jobs;
  55    int64_t max_data_size;             /* max data size */
  56    int64_t max_attr_size;
  57    int64_t data_size;                 /* current data size (all jobs running) */
  58    int64_t attr_size;
  59 };
  60
  61 static pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;
  62 spool_stats_t spool_stats;
  63
  64 /*
  65  * Header for data spool record */
  66 struct spool_hdr {
  67    int32_t  FirstIndex;               /* FirstIndex for buffer */
  68    int32_t  LastIndex;                /* LastIndex for buffer */
  69    uint32_t len;                      /* length of next buffer */
  70 };
  71
  72 enum {
  73    RB_EOT = 1,
  74    RB_ERROR,
  75    RB_OK
  76 };
  77
  78 void list_spool_stats(void sendit(const char *msg, int len, void *sarg), void *arg)
  79 {
  80    char ed1[30], ed2[30];
  81    POOL_MEM msg(PM_MESSAGE);
  82    int len;
  83
  84    len = Mmsg(msg, _("Spooling statistics:\n"));
  85
  86    if (spool_stats.data_jobs || spool_stats.max_data_size) {
  87       len = Mmsg(msg, _("Data spooling: %u active jobs, %s bytes; %u total jobs, %s max bytes/job.\n"),
  88          spool_stats.data_jobs, edit_uint64_with_commas(spool_stats.data_size, ed1),
  89          spool_stats.total_data_jobs,
  90          edit_uint64_with_commas(spool_stats.max_data_size, ed2));
  91
  92       sendit(msg.c_str(), len, arg);
  93    }
  94    if (spool_stats.attr_jobs || spool_stats.max_attr_size) {
  95       len = Mmsg(msg, _("Attr spooling: %u active jobs, %s bytes; %u total jobs, %s max bytes.\n"),
  96          spool_stats.attr_jobs, edit_uint64_with_commas(spool_stats.attr_size, ed1),
  97          spool_stats.total_attr_jobs,
  98          edit_uint64_with_commas(spool_stats.max_attr_size, ed2));
  99
 100       sendit(msg.c_str(), len, arg);
 101    }
 102 }
 103
 104 bool begin_data_spool(DCR *dcr)
 105 {
 106    bool stat = true;
 107    if (!dcr->dev->is_dvd() && dcr->jcr->spool_data) {
 108       Dmsg0(100, "Turning on data spooling\n");
 109       dcr->spool_data = true;
 110       stat = open_data_spool_file(dcr);
 111       if (stat) {
 112          dcr->spooling = true;
 113          Jmsg(dcr->jcr, M_INFO, 0, _("Spooling data ...\n"));
 114          P(mutex);
 115          spool_stats.data_jobs++;
 116          V(mutex);
 117       }
 118    }
 119    return stat;
 120 }
 121
 122 bool discard_data_spool(DCR *dcr)
 123 {
 124    if (dcr->spooling) {
 125       Dmsg0(100, "Data spooling discarded\n");
 126       return close_data_spool_file(dcr);
 127    }
 128    return true;
 129 }
 130
 131 bool commit_data_spool(DCR *dcr)
 132 {
 133    bool stat;
 134
 135    if (dcr->spooling) {
 136       Dmsg0(100, "Committing spooled data\n");
 137       stat = despool_data(dcr, true /*commit*/);
 138       if (!stat) {
 139          Dmsg1(100, _("Bad return from despool WroteVol=%d\n"), dcr->WroteVol);
 140          close_data_spool_file(dcr);
 141          return false;
 142       }
 143       return close_data_spool_file(dcr);
 144    }
 145    return true;
 146 }
 147
 148 static void make_unique_data_spool_filename(DCR *dcr, POOLMEM **name)
 149 {
 150    const char *dir;
 151    if (dcr->dev->device->spool_directory) {
 152       dir = dcr->dev->device->spool_directory;
 153    } else {
 154       dir = working_directory;
 155    }
 156    Mmsg(name, "%s/%s.data.%u.%s.%s.spool", dir, my_name, dcr->jcr->JobId,
 157         dcr->jcr->Job, dcr->device->hdr.name);
 158 }
 159
 160
 161 static bool open_data_spool_file(DCR *dcr)
 162 {
 163    POOLMEM *name  = get_pool_memory(PM_MESSAGE);
 164    int spool_fd;
 165
 166    make_unique_data_spool_filename(dcr, &name);
 167    if ((spool_fd = open(name, O_CREAT|O_TRUNC|O_RDWR|O_BINARY, 0640)) >= 0) {
 168       dcr->spool_fd = spool_fd;
 169       dcr->jcr->spool_attributes = true;
 170    } else {
 171       berrno be;
 172       Jmsg(dcr->jcr, M_FATAL, 0, _("Open data spool file %s failed: ERR=%s\n"), name,
 173            be.bstrerror());
 174       free_pool_memory(name);
 175       return false;
 176    }
 177    Dmsg1(100, "Created spool file: %s\n", name);
 178    free_pool_memory(name);
 179    return true;
 180 }
 181
 182 static bool close_data_spool_file(DCR *dcr)
 183 {
 184    POOLMEM *name  = get_pool_memory(PM_MESSAGE);
 185
 186    P(mutex);
 187    spool_stats.data_jobs--;
 188    spool_stats.total_data_jobs++;
 189    if (spool_stats.data_size < dcr->job_spool_size) {
 190       spool_stats.data_size = 0;
 191    } else {
 192       spool_stats.data_size -= dcr->job_spool_size;
 193    }
 194    dcr->job_spool_size = 0;
 195    V(mutex);
 196
 197    make_unique_data_spool_filename(dcr, &name);
 198    close(dcr->spool_fd);
 199    dcr->spool_fd = -1;
 200    dcr->spooling = false;
 201    unlink(name);
 202    Dmsg1(100, "Deleted spool file: %s\n", name);
 203    free_pool_memory(name);
 204    return true;
 205 }
 206
 207 static const char *spool_name = "*spool*";
 208
 209 /*
 210  * NB! This routine locks the device, but if committing will
 211  *     not unlock it. If not committing, it will be unlocked.
 212  */
 213 static bool despool_data(DCR *dcr, bool commit)
 214 {
 215    DEVICE *rdev;
 216    DCR *rdcr;
 217    bool ok = true;
 218    DEV_BLOCK *block;
 219    JCR *jcr = dcr->jcr;
 220    int stat;
 221    char ec1[50];
 222
 223    Dmsg0(100, "Despooling data\n");
 224    /*
 225     * Commit means that the job is done, so we commit, otherwise, we
 226     *  are despooling because of user spool size max or some error
 227     *  (e.g. filesystem full).
 228     */
 229    if (commit) {
 230       Jmsg(jcr, M_INFO, 0, _("Committing spooled data to Volume \"%s\". Despooling %s bytes ...\n"),
 231          jcr->dcr->VolumeName,
 232          edit_uint64_with_commas(jcr->dcr->job_spool_size, ec1));
 233       set_jcr_job_status(jcr, JS_DataCommitting);
 234    } else {
 235       Jmsg(jcr, M_INFO, 0, _("Writing spooled data to Volume. Despooling %s bytes ...\n"),
 236          edit_uint64_with_commas(jcr->dcr->job_spool_size, ec1));
 237       set_jcr_job_status(jcr, JS_DataDespooling);
 238    }
 239    set_jcr_job_status(jcr, JS_DataDespooling);
 240    dir_send_job_status(jcr);
 241    dcr->despool_wait = true;
 242    dcr->spooling = false;
 243    /*
 244     * We work with device blocked, but not locked so that
 245     *  other threads -- e.g. reservations can lock the device
 246     *  structure.
 247     */
 248    dcr->dblock(BST_DESPOOLING);
 249    dcr->despool_wait = false;
 250    dcr->despooling = true;
 251
 252    /*
 253     * This is really quite kludgy and should be fixed some time.
 254     * We create a dev structure to read from the spool file
 255     * in rdev and rdcr.
 256     */
 257    rdev = (DEVICE *)malloc(sizeof(DEVICE));
 258    memset(rdev, 0, sizeof(DEVICE));
 259    rdev->dev_name = get_memory(strlen(spool_name)+1);
 260    bstrncpy(rdev->dev_name, spool_name, sizeof(rdev->dev_name));
 261    rdev->errmsg = get_pool_memory(PM_EMSG);
 262    *rdev->errmsg = 0;
 263    rdev->max_block_size = dcr->dev->max_block_size;
 264    rdev->min_block_size = dcr->dev->min_block_size;
 265    rdev->device = dcr->dev->device;
 266    rdcr = new_dcr(jcr, NULL, rdev);
 267    rdcr->spool_fd = dcr->spool_fd;
 268    block = dcr->block;                /* save block */
 269    dcr->block = rdcr->block;          /* make read and write block the same */
 270
 271    Dmsg1(800, "read/write block size = %d\n", block->buf_len);
 272    lseek(rdcr->spool_fd, 0, SEEK_SET); /* rewind */
 273
 274 #if defined(HAVE_POSIX_FADVISE) && defined(POSIX_FADV_WILLNEED)
 275    posix_fadvise(rdcr->spool_fd, 0, 0, POSIX_FADV_WILLNEED);
 276 #endif
 277
 278    /* Add run time, to get current wait time */
 279    time_t despool_start = time(NULL) - jcr->run_time;
 280
 281    set_new_file_parameters(dcr);
 282
 283    for ( ; ok; ) {
 284       if (job_canceled(jcr)) {
 285          ok = false;
 286          break;
 287       }
 288       stat = read_block_from_spool_file(rdcr);
 289       if (stat == RB_EOT) {
 290          break;
 291       } else if (stat == RB_ERROR) {
 292          ok = false;
 293          break;
 294       }
 295       ok = write_block_to_device(dcr);
 296       if (!ok) {
 297          Jmsg2(jcr, M_FATAL, 0, _("Fatal append error on device %s: ERR=%s\n"),
 298                dcr->dev->print_name(), dcr->dev->bstrerror());
 299          Dmsg2(000, "Fatal append error on device %s: ERR=%s\n",
 300                dcr->dev->print_name(), dcr->dev->bstrerror());
 301       }
 302       Dmsg3(800, "Write block ok=%d FI=%d LI=%d\n", ok, block->FirstIndex, block->LastIndex);
 303    }
 304
 305    if (!dir_create_jobmedia_record(dcr)) {
 306       Jmsg(jcr, M_FATAL, 0, _("Could not create JobMedia record for Volume=\"%s\" Job=%s\n"),
 307          dcr->VolCatInfo.VolCatName, jcr->Job);
 308    }
 309    /* Set new file/block parameters for current dcr */
 310    set_new_file_parameters(dcr);
 311
 312    /* Subtracting run_time give us elapsed time - wait_time since we started despooling */
 313    time_t despool_elapsed = time(NULL) - despool_start - jcr->run_time;
 314
 315    if (despool_elapsed <= 0) {
 316       despool_elapsed = 1;
 317    }
 318
 319    Jmsg(dcr->jcr, M_INFO, 0, _("Despooling elapsed time = %02d:%02d:%02d, Transfer rate = %s bytes/second\n"),
 320          despool_elapsed / 3600, despool_elapsed % 3600 / 60, despool_elapsed % 60,
 321          edit_uint64_with_suffix(jcr->dcr->job_spool_size / despool_elapsed, ec1));
 322
 323    dcr->block = block;                /* reset block */
 324
 325    lseek(rdcr->spool_fd, 0, SEEK_SET); /* rewind */
 326    if (ftruncate(rdcr->spool_fd, 0) != 0) {
 327       berrno be;
 328       Jmsg(dcr->jcr, M_ERROR, 0, _("Ftruncate spool file failed: ERR=%s\n"),
 329          be.bstrerror());
 330       /* Note, try continuing despite ftruncate problem */
 331    }
 332
 333    P(mutex);
 334    if (spool_stats.data_size < dcr->job_spool_size) {
 335       spool_stats.data_size = 0;
 336    } else {
 337       spool_stats.data_size -= dcr->job_spool_size;
 338    }
 339    V(mutex);
 340    P(dcr->dev->spool_mutex);
 341    dcr->dev->spool_size -= dcr->job_spool_size;
 342    dcr->job_spool_size = 0;            /* zap size in input dcr */
 343    V(dcr->dev->spool_mutex);
 344    free_memory(rdev->dev_name);
 345    free_pool_memory(rdev->errmsg);
 346    /* Be careful to NULL the jcr and free rdev after free_dcr() */
 347    rdcr->jcr = NULL;
 348    rdcr->dev = NULL;
 349    free_dcr(rdcr);
 350    free(rdev);
 351    dcr->spooling = true;           /* turn on spooling again */
 352    dcr->despooling = false;
 353
 354    /*
 355     * We are done, so unblock the device, but if we have done a
 356     *  commit, leave it locked so that the job cleanup does not
 357     *  need to wait to release the device (no re-acquire of the lock).
 358     */
 359    dcr->dlock();
 360    unblock_device(dcr->dev);
 361    /* If doing a commit, leave the device locked -- unlocked in release_device() */
 362    if (!commit) {
 363       dcr->dunlock();
 364    }
 365    set_jcr_job_status(jcr, JS_Running);
 366    dir_send_job_status(jcr);
 367    return ok;
 368 }
 369
 370 /*
 371  * Read a block from the spool file
 372  *
 373  *  Returns RB_OK on success
 374  *          RB_EOT when file done
 375  *          RB_ERROR on error
 376  */
 377 static int read_block_from_spool_file(DCR *dcr)
 378 {
 379    uint32_t rlen;
 380    ssize_t stat;
 381    spool_hdr hdr;
 382    DEV_BLOCK *block = dcr->block;
 383
 384    rlen = sizeof(hdr);
 385    stat = read(dcr->spool_fd, (char *)&hdr, (size_t)rlen);
 386    if (stat == 0) {
 387       Dmsg0(100, "EOT on spool read.\n");
 388       return RB_EOT;
 389    } else if (stat != (ssize_t)rlen) {
 390       if (stat == -1) {
 391          berrno be;
 392          Jmsg(dcr->jcr, M_FATAL, 0, _("Spool header read error. ERR=%s\n"),
 393               be.bstrerror());
 394       } else {
 395          Pmsg2(000, _("Spool read error. Wanted %u bytes, got %d\n"), rlen, stat);
 396          Jmsg2(dcr->jcr, M_FATAL, 0, _("Spool header read error. Wanted %u bytes, got %d\n"), rlen, stat);
 397       }
 398       return RB_ERROR;
 399    }
 400    rlen = hdr.len;
 401    if (rlen > block->buf_len) {
 402       Pmsg2(000, _("Spool block too big. Max %u bytes, got %u\n"), block->buf_len, rlen);
 403       Jmsg2(dcr->jcr, M_FATAL, 0, _("Spool block too big. Max %u bytes, got %u\n"), block->buf_len, rlen);
 404       return RB_ERROR;
 405    }
 406    stat = read(dcr->spool_fd, (char *)block->buf, (size_t)rlen);
 407    if (stat != (ssize_t)rlen) {
 408       Pmsg2(000, _("Spool data read error. Wanted %u bytes, got %d\n"), rlen, stat);
 409       Jmsg2(dcr->jcr, M_FATAL, 0, _("Spool data read error. Wanted %u bytes, got %d\n"), rlen, stat);
 410       return RB_ERROR;
 411    }
 412    /* Setup write pointers */
 413    block->binbuf = rlen;
 414    block->bufp = block->buf + block->binbuf;
 415    block->FirstIndex = hdr.FirstIndex;
 416    block->LastIndex = hdr.LastIndex;
 417    block->VolSessionId = dcr->jcr->VolSessionId;
 418    block->VolSessionTime = dcr->jcr->VolSessionTime;
 419    Dmsg2(800, "Read block FI=%d LI=%d\n", block->FirstIndex, block->LastIndex);
 420    return RB_OK;
 421 }
 422
 423 /*
 424  * Write a block to the spool file
 425  *
 426  *  Returns: true on success or EOT
 427  *           false on hard error
 428  */
 429 bool write_block_to_spool_file(DCR *dcr)
 430 {
 431    uint32_t wlen, hlen;               /* length to write */
 432    bool despool = false;
 433    DEV_BLOCK *block = dcr->block;
 434
 435    ASSERT(block->binbuf == ((uint32_t) (block->bufp - block->buf)));
 436    if (block->binbuf <= WRITE_BLKHDR_LENGTH) {  /* Does block have data in it? */
 437       return true;
 438    }
 439
 440    hlen = sizeof(spool_hdr);
 441    wlen = block->binbuf;
 442    P(dcr->dev->spool_mutex);
 443    dcr->job_spool_size += hlen + wlen;
 444    dcr->dev->spool_size += hlen + wlen;
 445    if ((dcr->max_job_spool_size > 0 && dcr->job_spool_size >= dcr->max_job_spool_size) ||
 446        (dcr->dev->max_spool_size > 0 && dcr->dev->spool_size >= dcr->dev->max_spool_size)) {
 447       despool = true;
 448    }
 449    V(dcr->dev->spool_mutex);
 450    P(mutex);
 451    spool_stats.data_size += hlen + wlen;
 452    if (spool_stats.data_size > spool_stats.max_data_size) {
 453       spool_stats.max_data_size = spool_stats.data_size;
 454    }
 455    V(mutex);
 456    if (despool) {
 457 #ifdef xDEBUG
 458       char ec1[30], ec2[30], ec3[30], ec4[30];
 459       Dmsg4(100, "Despool in write_block_to_spool_file max_size=%s size=%s "
 460             "max_job_size=%s job_size=%s\n",
 461             edit_uint64_with_commas(dcr->max_job_spool_size, ec1),
 462             edit_uint64_with_commas(dcr->job_spool_size, ec2),
 463             edit_uint64_with_commas(dcr->dev->max_spool_size, ec3),
 464             edit_uint64_with_commas(dcr->dev->spool_size, ec4));
 465 #endif
 466       Jmsg(dcr->jcr, M_INFO, 0, _("User specified spool size reached.\n"));
 467       if (!despool_data(dcr, false)) {
 468          Pmsg0(000, _("Bad return from despool in write_block.\n"));
 469          return false;
 470       }
 471       /* Despooling cleared these variables so reset them */
 472       P(dcr->dev->spool_mutex);
 473       dcr->job_spool_size += hlen + wlen;
 474       dcr->dev->spool_size += hlen + wlen;
 475       V(dcr->dev->spool_mutex);
 476       Jmsg(dcr->jcr, M_INFO, 0, _("Spooling data again ...\n"));
 477    }
 478
 479
 480    if (!write_spool_header(dcr)) {
 481       return false;
 482    }
 483    if (!write_spool_data(dcr)) {
 484      return false;
 485    }
 486
 487    Dmsg2(800, "Wrote block FI=%d LI=%d\n", block->FirstIndex, block->LastIndex);
 488    empty_block(block);
 489    return true;
 490 }
 491
 492 static bool write_spool_header(DCR *dcr)
 493 {
 494    spool_hdr hdr;
 495    ssize_t stat;
 496    DEV_BLOCK *block = dcr->block;
 497
 498    hdr.FirstIndex = block->FirstIndex;
 499    hdr.LastIndex = block->LastIndex;
 500    hdr.len = block->binbuf;
 501
 502    /* Write header */
 503    for (int retry=0; retry<=1; retry++) {
 504       stat = write(dcr->spool_fd, (char*)&hdr, sizeof(hdr));
 505       if (stat == -1) {
 506          berrno be;
 507          Jmsg(dcr->jcr, M_FATAL, 0, _("Error writing header to spool file. ERR=%s\n"),
 508               be.bstrerror());
 509       }
 510       if (stat != (ssize_t)sizeof(hdr)) {
 511          /* If we wrote something, truncate it, then despool */
 512          if (stat != -1) {
 513 #if defined(HAVE_WIN32)
 514             boffset_t   pos = _lseeki64(dcr->spool_fd, (__int64)0, SEEK_CUR);
 515 #else
 516             boffset_t   pos = lseek(dcr->spool_fd, 0, SEEK_CUR);
 517 #endif
 518             if (ftruncate(dcr->spool_fd, pos - stat) != 0) {
 519                berrno be;
 520                Jmsg(dcr->jcr, M_ERROR, 0, _("Ftruncate spool file failed: ERR=%s\n"),
 521                   be.bstrerror());
 522               /* Note, try continuing despite ftruncate problem */
 523             }
 524          }
 525          if (!despool_data(dcr, false)) {
 526             Jmsg(dcr->jcr, M_FATAL, 0, _("Fatal despooling error."));
 527             return false;
 528          }
 529          continue;                    /* try again */
 530       }
 531       return true;
 532    }
 533    Jmsg(dcr->jcr, M_FATAL, 0, _("Retrying after header spooling error failed.\n"));
 534    return false;
 535 }
 536
 537 static bool write_spool_data(DCR *dcr)
 538 {
 539    ssize_t stat;
 540    DEV_BLOCK *block = dcr->block;
 541
 542    /* Write data */
 543    for (int retry=0; retry<=1; retry++) {
 544       stat = write(dcr->spool_fd, block->buf, (size_t)block->binbuf);
 545       if (stat == -1) {
 546          berrno be;
 547          Jmsg(dcr->jcr, M_FATAL, 0, _("Error writing data to spool file. ERR=%s\n"),
 548               be.bstrerror());
 549       }
 550       if (stat != (ssize_t)block->binbuf) {
 551          /*
 552           * If we wrote something, truncate it and the header, then despool
 553           */
 554          if (stat != -1) {
 555 #if defined(HAVE_WIN32)
 556             boffset_t   pos = _lseeki64(dcr->spool_fd, (__int64)0, SEEK_CUR);
 557 #else
 558             boffset_t   pos = lseek(dcr->spool_fd, 0, SEEK_CUR);
 559 #endif
 560             if (ftruncate(dcr->spool_fd, pos - stat - sizeof(spool_hdr)) != 0) {
 561                berrno be;
 562                Jmsg(dcr->jcr, M_ERROR, 0, _("Ftruncate spool file failed: ERR=%s\n"),
 563                   be.bstrerror());
 564                /* Note, try continuing despite ftruncate problem */
 565             }
 566          }
 567          if (!despool_data(dcr, false)) {
 568             Jmsg(dcr->jcr, M_FATAL, 0, _("Fatal despooling error."));
 569             return false;
 570          }
 571          if (!write_spool_header(dcr)) {
 572             return false;
 573          }
 574          continue;                    /* try again */
 575       }
 576       return true;
 577    }
 578    Jmsg(dcr->jcr, M_FATAL, 0, _("Retrying after data spooling error failed.\n"));
 579    return false;
 580 }
 581
 582
 583
 584 bool are_attributes_spooled(JCR *jcr)
 585 {
 586    return jcr->spool_attributes && jcr->dir_bsock->m_spool_fd;
 587 }
 588
 589 /*
 590  * Create spool file for attributes.
 591  *  This is done by "attaching" to the bsock, and when
 592  *  it is called, the output is written to a file.
 593  *  The actual spooling is turned on and off in
 594  *  append.c only during writing of the attributes.
 595  */
 596 bool begin_attribute_spool(JCR *jcr)
 597 {
 598    if (!jcr->no_attributes && jcr->spool_attributes) {
 599       return open_attr_spool_file(jcr, jcr->dir_bsock);
 600    }
 601    return true;
 602 }
 603
 604 bool discard_attribute_spool(JCR *jcr)
 605 {
 606    if (are_attributes_spooled(jcr)) {
 607       return close_attr_spool_file(jcr, jcr->dir_bsock);
 608    }
 609    return true;
 610 }
 611
 612 static void update_attr_spool_size(ssize_t size)
 613 {
 614    P(mutex);
 615    if (size > 0) {
 616      if ((spool_stats.attr_size - size) > 0) {
 617         spool_stats.attr_size -= size;
 618      } else {
 619         spool_stats.attr_size = 0;
 620      }
 621    }
 622    V(mutex);
 623 }
 624
 625 bool commit_attribute_spool(JCR *jcr)
 626 {
 627    boffset_t size;
 628    char ec1[30];
 629    char tbuf[100];
 630
 631    Dmsg1(100, "Commit attributes at %s\n", bstrftimes(tbuf, sizeof(tbuf),
 632          ( utime_t)time(NULL)));
 633    if (are_attributes_spooled(jcr)) {
 634       if (fseeko(jcr->dir_bsock->m_spool_fd, 0, SEEK_END) != 0) {
 635          berrno be;
 636          Jmsg(jcr, M_FATAL, 0, _("Fseek on attributes file failed: ERR=%s\n"),
 637               be.bstrerror());
 638          goto bail_out;
 639       }
 640       size = ftello(jcr->dir_bsock->m_spool_fd);
 641       if (size < 0) {
 642          berrno be;
 643          Jmsg(jcr, M_FATAL, 0, _("Fseek on attributes file failed: ERR=%s\n"),
 644               be.bstrerror());
 645          goto bail_out;
 646       }
 647       P(mutex);
 648       if (spool_stats.attr_size + size > spool_stats.max_attr_size) {
 649          spool_stats.max_attr_size = spool_stats.attr_size + size;
 650       }
 651       spool_stats.attr_size += size;
 652       V(mutex);
 653       set_jcr_job_status(jcr, JS_AttrDespooling);
 654       dir_send_job_status(jcr);
 655       Jmsg(jcr, M_INFO, 0, _("Sending spooled attrs to the Director. Despooling %s bytes ...\n"),
 656             edit_uint64_with_commas(size, ec1));
 657       jcr->dir_bsock->despool(update_attr_spool_size, size);
 658       return close_attr_spool_file(jcr, jcr->dir_bsock);
 659    }
 660    return true;
 661
 662 bail_out:
 663    close_attr_spool_file(jcr, jcr->dir_bsock);
 664    return false;
 665 }
 666
 667 static void make_unique_spool_filename(JCR *jcr, POOLMEM **name, int fd)
 668 {
 669    Mmsg(name, "%s/%s.attr.%s.%d.spool", working_directory, my_name,
 670       jcr->Job, fd);
 671 }
 672
 673
 674 bool open_attr_spool_file(JCR *jcr, BSOCK *bs)
 675 {
 676    POOLMEM *name  = get_pool_memory(PM_MESSAGE);
 677
 678    make_unique_spool_filename(jcr, &name, bs->m_fd);
 679    bs->m_spool_fd = fopen(name, "w+b");
 680    if (!bs->m_spool_fd) {
 681       berrno be;
 682       Jmsg(jcr, M_FATAL, 0, _("fopen attr spool file %s failed: ERR=%s\n"), name,
 683            be.bstrerror());
 684       free_pool_memory(name);
 685       return false;
 686    }
 687    P(mutex);
 688    spool_stats.attr_jobs++;
 689    V(mutex);
 690    free_pool_memory(name);
 691    return true;
 692 }
 693
 694 bool close_attr_spool_file(JCR *jcr, BSOCK *bs)
 695 {
 696    POOLMEM *name;
 697
 698    char tbuf[100];
 699
 700    Dmsg1(100, "Close attr spool file at %s\n", bstrftimes(tbuf, sizeof(tbuf),
 701          ( utime_t)time(NULL)));
 702    if (!bs->m_spool_fd) {
 703       return true;
 704    }
 705    name = get_pool_memory(PM_MESSAGE);
 706    P(mutex);
 707    spool_stats.attr_jobs--;
 708    spool_stats.total_attr_jobs++;
 709    V(mutex);
 710    make_unique_spool_filename(jcr, &name, bs->m_fd);
 711    fclose(bs->m_spool_fd);
 712    unlink(name);
 713    free_pool_memory(name);
 714    bs->m_spool_fd = NULL;
 715    bs->clear_spooling();
 716    return true;
 717 }