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 John Walker.
  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       }
 300       Dmsg3(800, "Write block ok=%d FI=%d LI=%d\n", ok, block->FirstIndex, block->LastIndex);
 301    }
 302
 303    if (!dir_create_jobmedia_record(dcr)) {
 304       Jmsg(jcr, M_FATAL, 0, _("Could not create JobMedia record for Volume=\"%s\" Job=%s\n"),
 305          dcr->VolCatInfo.VolCatName, jcr->Job);
 306    }
 307    /* Set new file/block parameters for current dcr */
 308    set_new_file_parameters(dcr);
 309
 310    /* Subtracting run_time give us elapsed time - wait_time since we started despooling */
 311    time_t despool_elapsed = time(NULL) - despool_start - jcr->run_time;
 312
 313    if (despool_elapsed <= 0) {
 314       despool_elapsed = 1;
 315    }
 316
 317    Jmsg(dcr->jcr, M_INFO, 0, _("Despooling elapsed time = %02d:%02d:%02d, Transfer rate = %s bytes/second\n"),
 318          despool_elapsed / 3600, despool_elapsed % 3600 / 60, despool_elapsed % 60,
 319          edit_uint64_with_suffix(jcr->dcr->job_spool_size / despool_elapsed, ec1));
 320
 321    dcr->block = block;                /* reset block */
 322
 323    lseek(rdcr->spool_fd, 0, SEEK_SET); /* rewind */
 324    if (ftruncate(rdcr->spool_fd, 0) != 0) {
 325       berrno be;
 326       Jmsg(dcr->jcr, M_ERROR, 0, _("Ftruncate spool file failed: ERR=%s\n"),
 327          be.bstrerror());
 328       /* Note, try continuing despite ftruncate problem */
 329    }
 330
 331    P(mutex);
 332    if (spool_stats.data_size < dcr->job_spool_size) {
 333       spool_stats.data_size = 0;
 334    } else {
 335       spool_stats.data_size -= dcr->job_spool_size;
 336    }
 337    V(mutex);
 338    P(dcr->dev->spool_mutex);
 339    dcr->dev->spool_size -= dcr->job_spool_size;
 340    dcr->job_spool_size = 0;            /* zap size in input dcr */
 341    V(dcr->dev->spool_mutex);
 342    free_memory(rdev->dev_name);
 343    free_pool_memory(rdev->errmsg);
 344    /* Be careful to NULL the jcr and free rdev after free_dcr() */
 345    rdcr->jcr = NULL;
 346    rdcr->dev = NULL;
 347    free_dcr(rdcr);
 348    free(rdev);
 349    dcr->spooling = true;           /* turn on spooling again */
 350    dcr->despooling = false;
 351
 352    /*
 353     * We are done, so unblock the device, but if we have done a
 354     *  commit, leave it locked so that the job cleanup does not
 355     *  need to wait to release the device (no re-acquire of the lock).
 356     */
 357    dcr->dlock();
 358    unblock_device(dcr->dev);
 359    /* If doing a commit, leave the device locked -- unlocked in release_device() */
 360    if (!commit) {
 361       dcr->dunlock();
 362    }
 363    set_jcr_job_status(jcr, JS_Running);
 364    dir_send_job_status(jcr);
 365    return ok;
 366 }
 367
 368 /*
 369  * Read a block from the spool file
 370  *
 371  *  Returns RB_OK on success
 372  *          RB_EOT when file done
 373  *          RB_ERROR on error
 374  */
 375 static int read_block_from_spool_file(DCR *dcr)
 376 {
 377    uint32_t rlen;
 378    ssize_t stat;
 379    spool_hdr hdr;
 380    DEV_BLOCK *block = dcr->block;
 381
 382    rlen = sizeof(hdr);
 383    stat = read(dcr->spool_fd, (char *)&hdr, (size_t)rlen);
 384    if (stat == 0) {
 385       Dmsg0(100, "EOT on spool read.\n");
 386       return RB_EOT;
 387    } else if (stat != (ssize_t)rlen) {
 388       if (stat == -1) {
 389          berrno be;
 390          Jmsg(dcr->jcr, M_FATAL, 0, _("Spool header read error. ERR=%s\n"),
 391               be.bstrerror());
 392       } else {
 393          Pmsg2(000, _("Spool read error. Wanted %u bytes, got %d\n"), rlen, stat);
 394          Jmsg2(dcr->jcr, M_FATAL, 0, _("Spool header read error. Wanted %u bytes, got %d\n"), rlen, stat);
 395       }
 396       return RB_ERROR;
 397    }
 398    rlen = hdr.len;
 399    if (rlen > block->buf_len) {
 400       Pmsg2(000, _("Spool block too big. Max %u bytes, got %u\n"), block->buf_len, rlen);
 401       Jmsg2(dcr->jcr, M_FATAL, 0, _("Spool block too big. Max %u bytes, got %u\n"), block->buf_len, rlen);
 402       return RB_ERROR;
 403    }
 404    stat = read(dcr->spool_fd, (char *)block->buf, (size_t)rlen);
 405    if (stat != (ssize_t)rlen) {
 406       Pmsg2(000, _("Spool data read error. Wanted %u bytes, got %d\n"), rlen, stat);
 407       Jmsg2(dcr->jcr, M_FATAL, 0, _("Spool data read error. Wanted %u bytes, got %d\n"), rlen, stat);
 408       return RB_ERROR;
 409    }
 410    /* Setup write pointers */
 411    block->binbuf = rlen;
 412    block->bufp = block->buf + block->binbuf;
 413    block->FirstIndex = hdr.FirstIndex;
 414    block->LastIndex = hdr.LastIndex;
 415    block->VolSessionId = dcr->jcr->VolSessionId;
 416    block->VolSessionTime = dcr->jcr->VolSessionTime;
 417    Dmsg2(800, "Read block FI=%d LI=%d\n", block->FirstIndex, block->LastIndex);
 418    return RB_OK;
 419 }
 420
 421 /*
 422  * Write a block to the spool file
 423  *
 424  *  Returns: true on success or EOT
 425  *           false on hard error
 426  */
 427 bool write_block_to_spool_file(DCR *dcr)
 428 {
 429    uint32_t wlen, hlen;               /* length to write */
 430    bool despool = false;
 431    DEV_BLOCK *block = dcr->block;
 432
 433    ASSERT(block->binbuf == ((uint32_t) (block->bufp - block->buf)));
 434    if (block->binbuf <= WRITE_BLKHDR_LENGTH) {  /* Does block have data in it? */
 435       return true;
 436    }
 437
 438    hlen = sizeof(spool_hdr);
 439    wlen = block->binbuf;
 440    P(dcr->dev->spool_mutex);
 441    dcr->job_spool_size += hlen + wlen;
 442    dcr->dev->spool_size += hlen + wlen;
 443    if ((dcr->max_job_spool_size > 0 && dcr->job_spool_size >= dcr->max_job_spool_size) ||
 444        (dcr->dev->max_spool_size > 0 && dcr->dev->spool_size >= dcr->dev->max_spool_size)) {
 445       despool = true;
 446    }
 447    V(dcr->dev->spool_mutex);
 448    P(mutex);
 449    spool_stats.data_size += hlen + wlen;
 450    if (spool_stats.data_size > spool_stats.max_data_size) {
 451       spool_stats.max_data_size = spool_stats.data_size;
 452    }
 453    V(mutex);
 454    if (despool) {
 455 #ifdef xDEBUG
 456       char ec1[30], ec2[30], ec3[30], ec4[30];
 457       Dmsg4(100, "Despool in write_block_to_spool_file max_size=%s size=%s "
 458             "max_job_size=%s job_size=%s\n",
 459             edit_uint64_with_commas(dcr->max_job_spool_size, ec1),
 460             edit_uint64_with_commas(dcr->job_spool_size, ec2),
 461             edit_uint64_with_commas(dcr->dev->max_spool_size, ec3),
 462             edit_uint64_with_commas(dcr->dev->spool_size, ec4));
 463 #endif
 464       Jmsg(dcr->jcr, M_INFO, 0, _("User specified spool size reached.\n"));
 465       if (!despool_data(dcr, false)) {
 466          Pmsg0(000, _("Bad return from despool in write_block.\n"));
 467          return false;
 468       }
 469       /* Despooling cleared these variables so reset them */
 470       P(dcr->dev->spool_mutex);
 471       dcr->job_spool_size += hlen + wlen;
 472       dcr->dev->spool_size += hlen + wlen;
 473       V(dcr->dev->spool_mutex);
 474       Jmsg(dcr->jcr, M_INFO, 0, _("Spooling data again ...\n"));
 475    }
 476
 477
 478    if (!write_spool_header(dcr)) {
 479       return false;
 480    }
 481    if (!write_spool_data(dcr)) {
 482      return false;
 483    }
 484
 485    Dmsg2(800, "Wrote block FI=%d LI=%d\n", block->FirstIndex, block->LastIndex);
 486    empty_block(block);
 487    return true;
 488 }
 489
 490 static bool write_spool_header(DCR *dcr)
 491 {
 492    spool_hdr hdr;
 493    ssize_t stat;
 494    DEV_BLOCK *block = dcr->block;
 495
 496    hdr.FirstIndex = block->FirstIndex;
 497    hdr.LastIndex = block->LastIndex;
 498    hdr.len = block->binbuf;
 499
 500    /* Write header */
 501    for (int retry=0; retry<=1; retry++) {
 502       stat = write(dcr->spool_fd, (char*)&hdr, sizeof(hdr));
 503       if (stat == -1) {
 504          berrno be;
 505          Jmsg(dcr->jcr, M_FATAL, 0, _("Error writing header to spool file. ERR=%s\n"),
 506               be.bstrerror());
 507       }
 508       if (stat != (ssize_t)sizeof(hdr)) {
 509          /* If we wrote something, truncate it, then despool */
 510          if (stat != -1) {
 511 #if defined(HAVE_WIN32)
 512             boffset_t   pos = _lseeki64(dcr->spool_fd, (__int64)0, SEEK_CUR);
 513 #else
 514             boffset_t   pos = lseek(dcr->spool_fd, (off_t)0, SEEK_CUR);
 515 #endif
 516             if (ftruncate(dcr->spool_fd, pos - stat) != 0) {
 517                berrno be;
 518                Jmsg(dcr->jcr, M_ERROR, 0, _("Ftruncate spool file failed: ERR=%s\n"),
 519                   be.bstrerror());
 520               /* Note, try continuing despite ftruncate problem */
 521             }
 522          }
 523          if (!despool_data(dcr, false)) {
 524             Jmsg(dcr->jcr, M_FATAL, 0, _("Fatal despooling error."));
 525             return false;
 526          }
 527          continue;                    /* try again */
 528       }
 529       return true;
 530    }
 531    Jmsg(dcr->jcr, M_FATAL, 0, _("Retrying after header spooling error failed.\n"));
 532    return false;
 533 }
 534
 535 static bool write_spool_data(DCR *dcr)
 536 {
 537    ssize_t stat;
 538    DEV_BLOCK *block = dcr->block;
 539
 540    /* Write data */
 541    for (int retry=0; retry<=1; retry++) {
 542       stat = write(dcr->spool_fd, block->buf, (size_t)block->binbuf);
 543       if (stat == -1) {
 544          berrno be;
 545          Jmsg(dcr->jcr, M_FATAL, 0, _("Error writing data to spool file. ERR=%s\n"),
 546               be.bstrerror());
 547       }
 548       if (stat != (ssize_t)block->binbuf) {
 549          /*
 550           * If we wrote something, truncate it and the header, then despool
 551           */
 552          if (stat != -1) {
 553 #if defined(HAVE_WIN32)
 554             boffset_t   pos = _lseeki64(dcr->spool_fd, (__int64)0, SEEK_CUR);
 555 #else
 556             boffset_t   pos = lseek(dcr->spool_fd, (off_t)0, SEEK_CUR);
 557 #endif
 558             if (ftruncate(dcr->spool_fd, pos - stat - sizeof(spool_hdr)) != 0) {
 559                berrno be;
 560                Jmsg(dcr->jcr, M_ERROR, 0, _("Ftruncate spool file failed: ERR=%s\n"),
 561                   be.bstrerror());
 562                /* Note, try continuing despite ftruncate problem */
 563             }
 564          }
 565          if (!despool_data(dcr, false)) {
 566             Jmsg(dcr->jcr, M_FATAL, 0, _("Fatal despooling error."));
 567             return false;
 568          }
 569          if (!write_spool_header(dcr)) {
 570             return false;
 571          }
 572          continue;                    /* try again */
 573       }
 574       return true;
 575    }
 576    Jmsg(dcr->jcr, M_FATAL, 0, _("Retrying after data spooling error failed.\n"));
 577    return false;
 578 }
 579
 580
 581
 582 bool are_attributes_spooled(JCR *jcr)
 583 {
 584    return jcr->spool_attributes && jcr->dir_bsock->m_spool_fd;
 585 }
 586
 587 /*
 588  * Create spool file for attributes.
 589  *  This is done by "attaching" to the bsock, and when
 590  *  it is called, the output is written to a file.
 591  *  The actual spooling is turned on and off in
 592  *  append.c only during writing of the attributes.
 593  */
 594 bool begin_attribute_spool(JCR *jcr)
 595 {
 596    if (!jcr->no_attributes && jcr->spool_attributes) {
 597       return open_attr_spool_file(jcr, jcr->dir_bsock);
 598    }
 599    return true;
 600 }
 601
 602 bool discard_attribute_spool(JCR *jcr)
 603 {
 604    if (are_attributes_spooled(jcr)) {
 605       return close_attr_spool_file(jcr, jcr->dir_bsock);
 606    }
 607    return true;
 608 }
 609
 610 static void update_attr_spool_size(ssize_t size)
 611 {
 612    P(mutex);
 613    if (size > 0) {
 614      if ((spool_stats.attr_size - size) > 0) {
 615         spool_stats.attr_size -= size;
 616      } else {
 617         spool_stats.attr_size = 0;
 618      }
 619    }
 620    V(mutex);
 621 }
 622
 623 bool commit_attribute_spool(JCR *jcr)
 624 {
 625    off_t size;
 626    char ec1[30];
 627    char tbuf[100];
 628
 629    Dmsg1(100, "Commit attributes at %s\n", bstrftimes(tbuf, sizeof(tbuf),
 630          ( utime_t)time(NULL)));
 631    if (are_attributes_spooled(jcr)) {
 632       if (fseeko(jcr->dir_bsock->m_spool_fd, 0, SEEK_END) != 0) {
 633          berrno be;
 634          Jmsg(jcr, M_FATAL, 0, _("Fseek on attributes file failed: ERR=%s\n"),
 635               be.bstrerror());
 636          goto bail_out;
 637       }
 638       size = ftello(jcr->dir_bsock->m_spool_fd);
 639       if (size < 0) {
 640          berrno be;
 641          Jmsg(jcr, M_FATAL, 0, _("Fseek on attributes file failed: ERR=%s\n"),
 642               be.bstrerror());
 643          goto bail_out;
 644       }
 645       P(mutex);
 646       if (spool_stats.attr_size + size > spool_stats.max_attr_size) {
 647          spool_stats.max_attr_size = spool_stats.attr_size + size;
 648       }
 649       spool_stats.attr_size += size;
 650       V(mutex);
 651       set_jcr_job_status(jcr, JS_AttrDespooling);
 652       dir_send_job_status(jcr);
 653       Jmsg(jcr, M_INFO, 0, _("Sending spooled attrs to the Director. Despooling %s bytes ...\n"),
 654             edit_uint64_with_commas(size, ec1));
 655       jcr->dir_bsock->despool(update_attr_spool_size, size);
 656       return close_attr_spool_file(jcr, jcr->dir_bsock);
 657    }
 658    return true;
 659
 660 bail_out:
 661    close_attr_spool_file(jcr, jcr->dir_bsock);
 662    return false;
 663 }
 664
 665 static void make_unique_spool_filename(JCR *jcr, POOLMEM **name, int fd)
 666 {
 667    Mmsg(name, "%s/%s.attr.%s.%d.spool", working_directory, my_name,
 668       jcr->Job, fd);
 669 }
 670
 671
 672 bool open_attr_spool_file(JCR *jcr, BSOCK *bs)
 673 {
 674    POOLMEM *name  = get_pool_memory(PM_MESSAGE);
 675
 676    make_unique_spool_filename(jcr, &name, bs->m_fd);
 677    bs->m_spool_fd = fopen(name, "w+b");
 678    if (!bs->m_spool_fd) {
 679       berrno be;
 680       Jmsg(jcr, M_FATAL, 0, _("fopen attr spool file %s failed: ERR=%s\n"), name,
 681            be.bstrerror());
 682       free_pool_memory(name);
 683       return false;
 684    }
 685    P(mutex);
 686    spool_stats.attr_jobs++;
 687    V(mutex);
 688    free_pool_memory(name);
 689    return true;
 690 }
 691
 692 bool close_attr_spool_file(JCR *jcr, BSOCK *bs)
 693 {
 694    POOLMEM *name;
 695
 696    char tbuf[100];
 697
 698    Dmsg1(100, "Close attr spool file at %s\n", bstrftimes(tbuf, sizeof(tbuf),
 699          ( utime_t)time(NULL)));
 700    if (!bs->m_spool_fd) {
 701       return true;
 702    }
 703    name = get_pool_memory(PM_MESSAGE);
 704    P(mutex);
 705    spool_stats.attr_jobs--;
 706    spool_stats.total_attr_jobs++;
 707    V(mutex);
 708    make_unique_spool_filename(jcr, &name, bs->m_fd);
 709    fclose(bs->m_spool_fd);
 710    unlink(name);
 711    free_pool_memory(name);
 712    bs->m_spool_fd = NULL;
 713    bs->clear_spooling();
 714    return true;
 715 }