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

   1 /*
   2  *  Spooling code
   3  *
   4  *      Kern Sibbald, March 2004
   5  *
   6  *  Version $Id$
   7  */
   8 /*
   9    Copyright (C) 2000-2004 Kern Sibbald and John Walker
  10
  11    This program is free software; you can redistribute it and/or
  12    modify it under the terms of the GNU General Public License as
  13    published by the Free Software Foundation; either version 2 of
  14    the License, or (at your option) any later version.
  15
  16    This program is distributed in the hope that it will be useful,
  17    but WITHOUT ANY WARRANTY; without even the implied warranty of
  18    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
  19    General Public License for more details.
  20
  21    You should have received a copy of the GNU General Public
  22    License along with this program; if not, write to the Free
  23    Software Foundation, Inc., 59 Temple Place - Suite 330, Boston,
  24    MA 02111-1307, USA.
  25
  26  */
  27
  28 #include "bacula.h"
  29 #include "stored.h"
  30
  31 /* Forward referenced subroutines */
  32 static void make_unique_data_spool_filename(JCR *jcr, POOLMEM **name);
  33 static int open_data_spool_file(JCR *jcr);
  34 static int close_data_spool_file(JCR *jcr);
  35 static bool despool_data(DCR *dcr);
  36 static int read_block_from_spool_file(DCR *dcr, DEV_BLOCK *block);
  37
  38 struct spool_hdr {
  39    int32_t  FirstIndex;
  40    int32_t  LastIndex;
  41    uint32_t len;
  42 };
  43
  44 enum {
  45    RB_EOT = 1,
  46    RB_ERROR,
  47    RB_OK
  48 };
  49
  50 bool begin_data_spool(JCR *jcr)
  51 {
  52    bool stat = true;
  53    if (jcr->spool_data) {
  54       Dmsg0(100, "Turning on data spooling\n");
  55       jcr->dcr->spool_data = true;
  56       stat = open_data_spool_file(jcr);
  57       if (stat) {
  58          jcr->dcr->spooling = true;
  59       }
  60    }
  61    return stat;
  62 }
  63
  64 bool discard_data_spool(JCR *jcr)
  65 {
  66    if (jcr->dcr->spooling) {
  67       Dmsg0(100, "Data spooling discarded\n");
  68       return close_data_spool_file(jcr);
  69    }
  70    return true;
  71 }
  72
  73 bool commit_data_spool(JCR *jcr)
  74 {
  75    bool stat;
  76    if (jcr->dcr->spooling) {
  77       Dmsg0(100, "Committing spooled data\n");
  78       stat = despool_data(jcr->dcr);
  79       if (!stat) {
  80          Dmsg1(000, "Bad return from despool WroteVol=%d\n", jcr->dcr->WroteVol);
  81          close_data_spool_file(jcr);
  82          return false;
  83       }
  84       return close_data_spool_file(jcr);
  85    }
  86    return true;
  87 }
  88
  89 static void make_unique_data_spool_filename(JCR *jcr, POOLMEM **name)
  90 {
  91    char *dir;
  92    if (jcr->dcr->dev->device->spool_directory) {
  93       dir = jcr->dcr->dev->device->spool_directory;
  94    } else {
  95       dir = working_directory;
  96    }
  97    Mmsg(name, "%s/%s.data.spool.%s.%s", dir, my_name, jcr->Job, jcr->device->hdr.name);
  98 }
  99
 100
 101 static int open_data_spool_file(JCR *jcr)
 102 {
 103    POOLMEM *name  = get_pool_memory(PM_MESSAGE);
 104    int spool_fd;
 105
 106    make_unique_data_spool_filename(jcr, &name);
 107    if ((spool_fd = open(name, O_CREAT|O_TRUNC|O_RDWR|O_BINARY, 0640)) >= 0) {
 108       jcr->dcr->spool_fd = spool_fd;
 109       jcr->spool_attributes = true;
 110    } else {
 111       Jmsg(jcr, M_ERROR, 0, "open data spool file %s failed: ERR=%s\n", name, strerror(errno));
 112       free_pool_memory(name);
 113       return 0;
 114     }
 115     Dmsg1(100, "Created spool file: %s\n", name);
 116     free_pool_memory(name);
 117     return 1;
 118 }
 119
 120 static int close_data_spool_file(JCR *jcr)
 121 {
 122     POOLMEM *name  = get_pool_memory(PM_MESSAGE);
 123
 124     make_unique_data_spool_filename(jcr, &name);
 125     close(jcr->dcr->spool_fd);
 126     jcr->dcr->spool_fd = -1;
 127     jcr->dcr->spooling = false;
 128     unlink(name);
 129     Dmsg1(100, "Deleted spool file: %s\n", name);
 130     free_pool_memory(name);
 131     return 1;
 132 }
 133
 134 static bool despool_data(DCR *dcr)
 135 {
 136    DEVICE *rdev;
 137    DCR *rdcr;
 138    bool ok = true;
 139    DEV_BLOCK *block;
 140    JCR *jcr = dcr->jcr;
 141    int stat;
 142
 143    Dmsg0(100, "Despooling data\n");
 144    dcr->spooling = false;
 145    lock_device(dcr->dev);
 146    dcr->dev_locked = true;
 147    /* Set up a dev structure to read */
 148    rdev = (DEVICE *)malloc(sizeof(DEVICE));
 149    memset(rdev, 0, sizeof(DEVICE));
 150    rdev->dev_name = get_memory(strlen("spool")+1);
 151    strcpy(rdev->dev_name, "spool");
 152    rdev->errmsg = get_pool_memory(PM_EMSG);
 153    *rdev->errmsg = 0;
 154    rdev->device = dcr->dev->device;
 155    rdcr = new_dcr(NULL, rdev);
 156    rdcr->spool_fd = dcr->spool_fd;
 157    rdcr->jcr = jcr;                   /* set a valid jcr */
 158    block = rdcr->block;
 159    lseek(rdcr->spool_fd, 0, SEEK_SET); /* rewind */
 160
 161    for ( ; ok; ) {
 162       if (job_canceled(jcr)) {
 163          ok = false;
 164          break;
 165       }
 166       stat = read_block_from_spool_file(rdcr, block);
 167       if (stat == RB_EOT) {
 168          break;
 169       } else if (stat == RB_ERROR) {
 170          ok = false;
 171          break;
 172       }
 173       ok = write_block_to_device(dcr, block);
 174       Dmsg3(100, "Write block ok=%d FI=%d LI=%d\n", ok, block->FirstIndex, block->LastIndex);
 175    }
 176
 177    lseek(rdcr->spool_fd, 0, SEEK_SET); /* rewind */
 178    if (ftruncate(rdcr->spool_fd, 0) != 0) {
 179       Dmsg1(000, "Bad return from ftruncate. ERR=%s\n", strerror(errno));
 180       ok = false;
 181    }
 182
 183    P(dcr->dev->spool_mutex);
 184    dcr->dev->spool_size -= dcr->spool_size;
 185    dcr->spool_size = 0;               /* zap size in input dcr */
 186    V(dcr->dev->spool_mutex);
 187
 188    free_memory(rdev->dev_name);
 189    free_pool_memory(rdev->errmsg);
 190    free(rdev);
 191    rdcr->jcr = NULL;
 192    free_dcr(rdcr);
 193    unlock_device(dcr->dev);
 194    dcr->dev_locked = false;
 195    dcr->spooling = true;           /* turn on spooling again */
 196    return ok;
 197 }
 198
 199 /*
 200  * Read a block from the spool file
 201  *
 202  *  Returns RB_OK on success
 203  *          RB_EOT when file done
 204  *          RB_ERROR on error
 205  */
 206 static int read_block_from_spool_file(DCR *dcr, DEV_BLOCK *block)
 207 {
 208    uint32_t rlen;
 209    ssize_t stat;
 210    spool_hdr hdr;
 211
 212    rlen = sizeof(hdr);
 213    stat = read(dcr->spool_fd, (char *)&hdr, (size_t)rlen);
 214    if (stat == 0) {
 215       Dmsg0(100, "EOT on spool read.\n");
 216       return RB_EOT;
 217    } else if (stat != (ssize_t)rlen) {
 218       if (stat == -1) {
 219          Jmsg(dcr->jcr, M_FATAL, 0, "Spool read error. ERR=%s\n", strerror(errno));
 220       } else {
 221          Dmsg2(000, "Spool read error. Wanted %u bytes, got %u\n", rlen, stat);
 222          Jmsg2(dcr->jcr, M_FATAL, 0, "Spool read error. Wanted %u bytes, got %u\n", rlen, stat);
 223       }
 224       return RB_ERROR;
 225    }
 226    rlen = hdr.len;
 227    if (rlen > block->buf_len) {
 228       Dmsg2(000, "Spool block too big. Max %u bytes, got %u\n", block->buf_len, rlen);
 229       Jmsg2(dcr->jcr, M_FATAL, 0, "Spool block too big. Max %u bytes, got %u\n", block->buf_len, rlen);
 230       return RB_ERROR;
 231    }
 232    stat = read(dcr->spool_fd, (char *)block->buf, (size_t)rlen);
 233    if (stat != (ssize_t)rlen) {
 234       Dmsg2(000, "Spool read error. Wanted %u bytes, got %u\n", rlen, stat);
 235       Jmsg2(dcr->jcr, M_FATAL, 0, "Spool read error. Wanted %u bytes, got %u\n", rlen, stat);
 236       return RB_ERROR;
 237    }
 238    /* Setup write pointers */
 239    block->binbuf = rlen;
 240    block->bufp = block->buf + block->binbuf;
 241    block->FirstIndex = hdr.FirstIndex;
 242    block->LastIndex = hdr.LastIndex;
 243    block->VolSessionId = dcr->jcr->VolSessionId;
 244    block->VolSessionTime = dcr->jcr->VolSessionTime;
 245    Dmsg2(100, "Read block FI=%d LI=%d\n", block->FirstIndex, block->LastIndex);
 246    return RB_OK;
 247 }
 248
 249 /*
 250  * Write a block to the spool file
 251  *
 252  *  Returns: true on success or EOT
 253  *           false on hard error
 254  */
 255 bool write_block_to_spool_file(DCR *dcr, DEV_BLOCK *block)
 256 {
 257    ssize_t stat = 0;
 258    uint32_t wlen, hlen;               /* length to write */
 259    int retry = 0;
 260    spool_hdr hdr;
 261    bool despool = false;
 262
 263    ASSERT(block->binbuf == ((uint32_t) (block->bufp - block->buf)));
 264    if (block->binbuf <= WRITE_BLKHDR_LENGTH) {  /* Does block have data in it? */
 265       Dmsg0(100, "return write_block_to_dev no data to write\n");
 266       return true;
 267    }
 268
 269    hlen = sizeof(hdr);
 270    wlen = block->binbuf;
 271    P(dcr->dev->spool_mutex);
 272    dcr->spool_size += hlen + wlen;
 273    dcr->dev->spool_size += hlen + wlen;
 274    if ((dcr->max_spool_size > 0 && dcr->spool_size >= dcr->max_spool_size) ||
 275        (dcr->dev->max_spool_size > 0 && dcr->dev->spool_size >= dcr->dev->max_spool_size)) {
 276       despool = true;
 277    }
 278    V(dcr->dev->spool_mutex);
 279    if (despool) {
 280       char ec1[30], ec2[30], ec3[30], ec4[30];
 281       Dmsg4(100, "Despool in write_block_to_spool_file max_size=%s size=%s "
 282             "max_job_size=%s job_size=%s\n",
 283             edit_uint64_with_commas(dcr->max_spool_size, ec1),
 284             edit_uint64_with_commas(dcr->spool_size, ec2),
 285             edit_uint64_with_commas(dcr->dev->max_spool_size, ec3),
 286             edit_uint64_with_commas(dcr->dev->spool_size, ec4));
 287       despool = false;
 288       if (!despool_data(dcr)) {
 289          Dmsg0(000, "Bad return from despool in write_block.\n");
 290          return false;
 291       }
 292       P(dcr->dev->spool_mutex);
 293       dcr->spool_size += hlen + wlen;
 294       dcr->dev->spool_size += hlen + wlen;
 295       V(dcr->dev->spool_mutex);
 296    }
 297
 298    hdr.FirstIndex = block->FirstIndex;
 299    hdr.LastIndex = block->LastIndex;
 300    hdr.len = block->binbuf;
 301
 302    /* Write header */
 303    for ( ;; ) {
 304       stat = write(dcr->spool_fd, (char*)&hdr, (size_t)hlen);
 305       if (stat != (ssize_t)hlen) {
 306          if (!despool_data(dcr)) {
 307             return false;
 308          }
 309          if (retry++ > 1) {
 310             return false;
 311          }
 312          continue;
 313       }
 314       break;
 315    }
 316
 317    /* Write data */
 318    for ( ;; ) {
 319       stat = write(dcr->spool_fd, block->buf, (size_t)wlen);
 320       if (stat != (ssize_t)wlen) {
 321          if (!despool_data(dcr)) {
 322             return false;
 323          }
 324          if (retry++ > 1) {
 325             return false;
 326          }
 327          continue;
 328       }
 329       break;
 330    }
 331    Dmsg2(100, "Wrote block FI=%d LI=%d\n", block->FirstIndex, block->LastIndex);
 332
 333    empty_block(block);
 334    return true;
 335 }
 336
 337
 338 bool are_attributes_spooled(JCR *jcr)
 339 {
 340    return jcr->spool_attributes && jcr->dir_bsock->spool_fd;
 341 }
 342
 343 /*
 344  * Create spool file for attributes.
 345  *  This is done by "attaching" to the bsock, and when
 346  *  it is called, the output is written to a file.
 347  *  The actual spooling is turned on and off in
 348  *  append.c only during writing of the attributes.
 349  */
 350 bool begin_attribute_spool(JCR *jcr)
 351 {
 352    if (!jcr->no_attributes && jcr->spool_attributes) {
 353       return open_spool_file(jcr, jcr->dir_bsock);
 354    }
 355    return true;
 356 }
 357
 358 bool discard_attribute_spool(JCR *jcr)
 359 {
 360    if (are_attributes_spooled(jcr)) {
 361       return close_spool_file(jcr, jcr->dir_bsock);
 362    }
 363    return true;
 364 }
 365
 366 bool commit_attribute_spool(JCR *jcr)
 367 {
 368    if (are_attributes_spooled(jcr)) {
 369       bnet_despool_to_bsock(jcr->dir_bsock);
 370       return close_spool_file(jcr, jcr->dir_bsock);
 371    }
 372    return true;
 373 }
 374
 375 static void make_unique_spool_filename(JCR *jcr, POOLMEM **name, int fd)
 376 {
 377    Mmsg(name, "%s/%s.spool.%s.%d", working_directory, my_name,
 378       jcr->Job, fd);
 379 }
 380
 381 bool open_spool_file(JCR *jcr, BSOCK *bs)
 382 {
 383     POOLMEM *name  = get_pool_memory(PM_MESSAGE);
 384
 385     make_unique_spool_filename(jcr, &name, bs->fd);
 386     bs->spool_fd = fopen(mp_chr(name), "w+");
 387     if (!bs->spool_fd) {
 388        Jmsg(jcr, M_ERROR, 0, "fopen spool file %s failed: ERR=%s\n", name, strerror(errno));
 389        free_pool_memory(name);
 390        return false;
 391     }
 392     free_pool_memory(name);
 393     return true;
 394 }
 395
 396 bool close_spool_file(JCR *jcr, BSOCK *bs)
 397 {
 398     POOLMEM *name  = get_pool_memory(PM_MESSAGE);
 399
 400     make_unique_spool_filename(jcr, &name, bs->fd);
 401     fclose(bs->spool_fd);
 402     unlink(mp_chr(name));
 403     free_pool_memory(name);
 404     bs->spool_fd = NULL;
 405     bs->spool = false;
 406     return true;
 407 }