2 Bacula® - The Network Backup Solution
4 Copyright (C) 2004-2007 Free Software Foundation Europe e.V.
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
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.
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
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.
31 * Kern Sibbald, March 2004
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);
50 struct spool_stats_t {
51 uint32_t data_jobs; /* current jobs spooling data */
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) */
61 static pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;
62 spool_stats_t spool_stats;
65 * Header for data spool record */
67 int32_t FirstIndex; /* FirstIndex for buffer */
68 int32_t LastIndex; /* LastIndex for buffer */
69 uint32_t len; /* length of next buffer */
78 void list_spool_stats(void sendit(const char *msg, int len, void *sarg), void *arg)
80 char ed1[30], ed2[30];
81 POOL_MEM msg(PM_MESSAGE);
84 len = Mmsg(msg, _("Spooling statistics:\n"));
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));
92 sendit(msg.c_str(), len, arg);
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));
100 sendit(msg.c_str(), len, arg);
102 len = Mmsg(msg, "====\n");
103 sendit(msg.c_str(), len, arg);
106 bool begin_data_spool(DCR *dcr)
109 if (!dcr->dev->is_dvd() && dcr->jcr->spool_data) {
110 Dmsg0(100, "Turning on data spooling\n");
111 dcr->spool_data = true;
112 stat = open_data_spool_file(dcr);
114 dcr->spooling = true;
115 Jmsg(dcr->jcr, M_INFO, 0, _("Spooling data ...\n"));
117 spool_stats.data_jobs++;
124 bool discard_data_spool(DCR *dcr)
127 Dmsg0(100, "Data spooling discarded\n");
128 return close_data_spool_file(dcr);
133 bool commit_data_spool(DCR *dcr)
138 Dmsg0(100, "Committing spooled data\n");
139 stat = despool_data(dcr, true /*commit*/);
141 Dmsg1(100, _("Bad return from despool WroteVol=%d\n"), dcr->WroteVol);
142 close_data_spool_file(dcr);
145 return close_data_spool_file(dcr);
150 static void make_unique_data_spool_filename(DCR *dcr, POOLMEM **name)
153 if (dcr->dev->device->spool_directory) {
154 dir = dcr->dev->device->spool_directory;
156 dir = working_directory;
158 Mmsg(name, "%s/%s.data.%u.%s.%s.spool", dir, my_name, dcr->jcr->JobId,
159 dcr->jcr->Job, dcr->device->hdr.name);
163 static bool open_data_spool_file(DCR *dcr)
165 POOLMEM *name = get_pool_memory(PM_MESSAGE);
168 make_unique_data_spool_filename(dcr, &name);
169 if ((spool_fd = open(name, O_CREAT|O_TRUNC|O_RDWR|O_BINARY, 0640)) >= 0) {
170 dcr->spool_fd = spool_fd;
171 dcr->jcr->spool_attributes = true;
174 Jmsg(dcr->jcr, M_FATAL, 0, _("Open data spool file %s failed: ERR=%s\n"), name,
176 free_pool_memory(name);
179 Dmsg1(100, "Created spool file: %s\n", name);
180 free_pool_memory(name);
184 static bool close_data_spool_file(DCR *dcr)
186 POOLMEM *name = get_pool_memory(PM_MESSAGE);
189 spool_stats.data_jobs--;
190 spool_stats.total_data_jobs++;
191 if (spool_stats.data_size < dcr->job_spool_size) {
192 spool_stats.data_size = 0;
194 spool_stats.data_size -= dcr->job_spool_size;
196 dcr->job_spool_size = 0;
199 make_unique_data_spool_filename(dcr, &name);
200 close(dcr->spool_fd);
202 dcr->spooling = false;
204 Dmsg1(100, "Deleted spool file: %s\n", name);
205 free_pool_memory(name);
209 static const char *spool_name = "*spool*";
212 * NB! This routine locks the device, but if committing will
213 * not unlock it. If not committing, it will be unlocked.
215 static bool despool_data(DCR *dcr, bool commit)
225 Dmsg0(100, "Despooling data\n");
227 * Commit means that the job is done, so we commit, otherwise, we
228 * are despooling because of user spool size max or some error
229 * (e.g. filesystem full).
232 Jmsg(jcr, M_INFO, 0, _("Committing spooled data to Volume \"%s\". Despooling %s bytes ...\n"),
233 jcr->dcr->VolumeName,
234 edit_uint64_with_commas(jcr->dcr->job_spool_size, ec1));
236 Jmsg(jcr, M_INFO, 0, _("Writing spooled data to Volume. Despooling %s bytes ...\n"),
237 edit_uint64_with_commas(jcr->dcr->job_spool_size, ec1));
239 dcr->despool_wait = true;
240 dcr->spooling = false;
242 * We work with device blocked, but not locked so that
243 * other threads -- e.g. reservations can lock the device
246 dcr->dblock(BST_DESPOOLING);
247 dcr->despool_wait = false;
248 dcr->despooling = true;
251 * This is really quite kludgy and should be fixed some time.
252 * We create a dev structure to read from the spool file
255 rdev = (DEVICE *)malloc(sizeof(DEVICE));
256 memset(rdev, 0, sizeof(DEVICE));
257 rdev->dev_name = get_memory(strlen(spool_name)+1);
258 bstrncpy(rdev->dev_name, spool_name, sizeof(rdev->dev_name));
259 rdev->errmsg = get_pool_memory(PM_EMSG);
261 rdev->max_block_size = dcr->dev->max_block_size;
262 rdev->min_block_size = dcr->dev->min_block_size;
263 rdev->device = dcr->dev->device;
264 rdcr = new_dcr(jcr, NULL, rdev);
265 rdcr->spool_fd = dcr->spool_fd;
266 block = dcr->block; /* save block */
267 dcr->block = rdcr->block; /* make read and write block the same */
269 Dmsg1(800, "read/write block size = %d\n", block->buf_len);
270 lseek(rdcr->spool_fd, 0, SEEK_SET); /* rewind */
272 #if defined(HAVE_POSIX_FADVISE) && defined(POSIX_FADV_WILLNEED)
273 posix_fadvise(rdcr->spool_fd, 0, 0, POSIX_FADV_WILLNEED);
276 /* Add run time, to get current wait time */
277 time_t despool_start = time(NULL) - jcr->run_time;
279 set_new_file_parameters(dcr);
282 if (job_canceled(jcr)) {
286 stat = read_block_from_spool_file(rdcr);
287 if (stat == RB_EOT) {
289 } else if (stat == RB_ERROR) {
293 ok = write_block_to_device(dcr);
295 Jmsg2(jcr, M_FATAL, 0, _("Fatal append error on device %s: ERR=%s\n"),
296 dcr->dev->print_name(), dcr->dev->bstrerror());
298 Dmsg3(800, "Write block ok=%d FI=%d LI=%d\n", ok, block->FirstIndex, block->LastIndex);
301 if (!dir_create_jobmedia_record(dcr)) {
302 Jmsg(jcr, M_FATAL, 0, _("Could not create JobMedia record for Volume=\"%s\" Job=%s\n"),
303 dcr->VolCatInfo.VolCatName, jcr->Job);
305 /* Set new file/block parameters for current dcr */
306 set_new_file_parameters(dcr);
308 /* Subtracting run_time give us elapsed time - wait_time since we started despooling */
309 time_t despool_elapsed = time(NULL) - despool_start - jcr->run_time;
311 if (despool_elapsed <= 0) {
315 Jmsg(dcr->jcr, M_INFO, 0, _("Despooling elapsed time = %02d:%02d:%02d, Transfer rate = %s bytes/second\n"),
316 despool_elapsed / 3600, despool_elapsed % 3600 / 60, despool_elapsed % 60,
317 edit_uint64_with_suffix(jcr->dcr->job_spool_size / despool_elapsed, ec1));
319 dcr->block = block; /* reset block */
321 lseek(rdcr->spool_fd, 0, SEEK_SET); /* rewind */
322 if (ftruncate(rdcr->spool_fd, 0) != 0) {
324 Jmsg(dcr->jcr, M_ERROR, 0, _("Ftruncate spool file failed: ERR=%s\n"),
326 /* Note, try continuing despite ftruncate problem */
330 if (spool_stats.data_size < dcr->job_spool_size) {
331 spool_stats.data_size = 0;
333 spool_stats.data_size -= dcr->job_spool_size;
336 P(dcr->dev->spool_mutex);
337 dcr->dev->spool_size -= dcr->job_spool_size;
338 dcr->job_spool_size = 0; /* zap size in input dcr */
339 V(dcr->dev->spool_mutex);
340 free_memory(rdev->dev_name);
341 free_pool_memory(rdev->errmsg);
342 /* Be careful to NULL the jcr and free rdev after free_dcr() */
347 dcr->spooling = true; /* turn on spooling again */
348 dcr->despooling = false;
350 * We are done, so unblock the device, but if we have done a
351 * commit, leave it locked so that the job cleanup does not
352 * need to wait to release the device (no re-acquire of the lock).
355 unblock_device(dcr->dev);
356 /* If doing a commit, leave the device locked -- unlocked in release_device() */
364 * Read a block from the spool file
366 * Returns RB_OK on success
367 * RB_EOT when file done
370 static int read_block_from_spool_file(DCR *dcr)
375 DEV_BLOCK *block = dcr->block;
378 stat = read(dcr->spool_fd, (char *)&hdr, (size_t)rlen);
380 Dmsg0(100, "EOT on spool read.\n");
382 } else if (stat != (ssize_t)rlen) {
385 Jmsg(dcr->jcr, M_FATAL, 0, _("Spool header read error. ERR=%s\n"),
388 Pmsg2(000, _("Spool read error. Wanted %u bytes, got %d\n"), rlen, stat);
389 Jmsg2(dcr->jcr, M_FATAL, 0, _("Spool header read error. Wanted %u bytes, got %d\n"), rlen, stat);
394 if (rlen > block->buf_len) {
395 Pmsg2(000, _("Spool block too big. Max %u bytes, got %u\n"), block->buf_len, rlen);
396 Jmsg2(dcr->jcr, M_FATAL, 0, _("Spool block too big. Max %u bytes, got %u\n"), block->buf_len, rlen);
399 stat = read(dcr->spool_fd, (char *)block->buf, (size_t)rlen);
400 if (stat != (ssize_t)rlen) {
401 Pmsg2(000, _("Spool data read error. Wanted %u bytes, got %d\n"), rlen, stat);
402 Jmsg2(dcr->jcr, M_FATAL, 0, _("Spool data read error. Wanted %u bytes, got %d\n"), rlen, stat);
405 /* Setup write pointers */
406 block->binbuf = rlen;
407 block->bufp = block->buf + block->binbuf;
408 block->FirstIndex = hdr.FirstIndex;
409 block->LastIndex = hdr.LastIndex;
410 block->VolSessionId = dcr->jcr->VolSessionId;
411 block->VolSessionTime = dcr->jcr->VolSessionTime;
412 Dmsg2(800, "Read block FI=%d LI=%d\n", block->FirstIndex, block->LastIndex);
417 * Write a block to the spool file
419 * Returns: true on success or EOT
420 * false on hard error
422 bool write_block_to_spool_file(DCR *dcr)
424 uint32_t wlen, hlen; /* length to write */
425 bool despool = false;
426 DEV_BLOCK *block = dcr->block;
428 ASSERT(block->binbuf == ((uint32_t) (block->bufp - block->buf)));
429 if (block->binbuf <= WRITE_BLKHDR_LENGTH) { /* Does block have data in it? */
433 hlen = sizeof(spool_hdr);
434 wlen = block->binbuf;
435 P(dcr->dev->spool_mutex);
436 dcr->job_spool_size += hlen + wlen;
437 dcr->dev->spool_size += hlen + wlen;
438 if ((dcr->max_job_spool_size > 0 && dcr->job_spool_size >= dcr->max_job_spool_size) ||
439 (dcr->dev->max_spool_size > 0 && dcr->dev->spool_size >= dcr->dev->max_spool_size)) {
442 V(dcr->dev->spool_mutex);
444 spool_stats.data_size += hlen + wlen;
445 if (spool_stats.data_size > spool_stats.max_data_size) {
446 spool_stats.max_data_size = spool_stats.data_size;
451 char ec1[30], ec2[30], ec3[30], ec4[30];
452 Dmsg4(100, "Despool in write_block_to_spool_file max_size=%s size=%s "
453 "max_job_size=%s job_size=%s\n",
454 edit_uint64_with_commas(dcr->max_job_spool_size, ec1),
455 edit_uint64_with_commas(dcr->job_spool_size, ec2),
456 edit_uint64_with_commas(dcr->dev->max_spool_size, ec3),
457 edit_uint64_with_commas(dcr->dev->spool_size, ec4));
459 Jmsg(dcr->jcr, M_INFO, 0, _("User specified spool size reached.\n"));
460 if (!despool_data(dcr, false)) {
461 Pmsg0(000, _("Bad return from despool in write_block.\n"));
464 /* Despooling cleared these variables so reset them */
465 P(dcr->dev->spool_mutex);
466 dcr->job_spool_size += hlen + wlen;
467 dcr->dev->spool_size += hlen + wlen;
468 V(dcr->dev->spool_mutex);
469 Jmsg(dcr->jcr, M_INFO, 0, _("Spooling data again ...\n"));
473 if (!write_spool_header(dcr)) {
476 if (!write_spool_data(dcr)) {
480 Dmsg2(800, "Wrote block FI=%d LI=%d\n", block->FirstIndex, block->LastIndex);
485 static bool write_spool_header(DCR *dcr)
489 DEV_BLOCK *block = dcr->block;
491 hdr.FirstIndex = block->FirstIndex;
492 hdr.LastIndex = block->LastIndex;
493 hdr.len = block->binbuf;
496 for (int retry=0; retry<=1; retry++) {
497 stat = write(dcr->spool_fd, (char*)&hdr, sizeof(hdr));
500 Jmsg(dcr->jcr, M_FATAL, 0, _("Error writing header to spool file. ERR=%s\n"),
503 if (stat != (ssize_t)sizeof(hdr)) {
504 /* If we wrote something, truncate it, then despool */
506 #if defined(HAVE_WIN32)
507 boffset_t pos = _lseeki64(dcr->spool_fd, (__int64)0, SEEK_CUR);
509 boffset_t pos = lseek(dcr->spool_fd, (off_t)0, SEEK_CUR);
511 if (ftruncate(dcr->spool_fd, pos - stat) != 0) {
513 Jmsg(dcr->jcr, M_ERROR, 0, _("Ftruncate spool file failed: ERR=%s\n"),
515 /* Note, try continuing despite ftruncate problem */
518 if (!despool_data(dcr, false)) {
519 Jmsg(dcr->jcr, M_FATAL, 0, _("Fatal despooling error."));
522 continue; /* try again */
526 Jmsg(dcr->jcr, M_FATAL, 0, _("Retrying after header spooling error failed.\n"));
530 static bool write_spool_data(DCR *dcr)
533 DEV_BLOCK *block = dcr->block;
536 for (int retry=0; retry<=1; retry++) {
537 stat = write(dcr->spool_fd, block->buf, (size_t)block->binbuf);
540 Jmsg(dcr->jcr, M_FATAL, 0, _("Error writing data to spool file. ERR=%s\n"),
543 if (stat != (ssize_t)block->binbuf) {
545 * If we wrote something, truncate it and the header, then despool
548 #if defined(HAVE_WIN32)
549 boffset_t pos = _lseeki64(dcr->spool_fd, (__int64)0, SEEK_CUR);
551 boffset_t pos = lseek(dcr->spool_fd, (off_t)0, SEEK_CUR);
553 if (ftruncate(dcr->spool_fd, pos - stat - sizeof(spool_hdr)) != 0) {
555 Jmsg(dcr->jcr, M_ERROR, 0, _("Ftruncate spool file failed: ERR=%s\n"),
557 /* Note, try continuing despite ftruncate problem */
560 if (!despool_data(dcr, false)) {
561 Jmsg(dcr->jcr, M_FATAL, 0, _("Fatal despooling error."));
564 if (!write_spool_header(dcr)) {
567 continue; /* try again */
571 Jmsg(dcr->jcr, M_FATAL, 0, _("Retrying after data spooling error failed.\n"));
577 bool are_attributes_spooled(JCR *jcr)
579 return jcr->spool_attributes && jcr->dir_bsock->m_spool_fd;
583 * Create spool file for attributes.
584 * This is done by "attaching" to the bsock, and when
585 * it is called, the output is written to a file.
586 * The actual spooling is turned on and off in
587 * append.c only during writing of the attributes.
589 bool begin_attribute_spool(JCR *jcr)
591 if (!jcr->no_attributes && jcr->spool_attributes) {
592 return open_attr_spool_file(jcr, jcr->dir_bsock);
597 bool discard_attribute_spool(JCR *jcr)
599 if (are_attributes_spooled(jcr)) {
600 return close_attr_spool_file(jcr, jcr->dir_bsock);
605 static void update_attr_spool_size(ssize_t size)
609 if ((spool_stats.attr_size - size) > 0) {
610 spool_stats.attr_size -= size;
612 spool_stats.attr_size = 0;
618 bool commit_attribute_spool(JCR *jcr)
623 if (are_attributes_spooled(jcr)) {
624 if (fseeko(jcr->dir_bsock->m_spool_fd, 0, SEEK_END) != 0) {
626 Jmsg(jcr, M_FATAL, 0, _("Fseek on attributes file failed: ERR=%s\n"),
630 size = ftello(jcr->dir_bsock->m_spool_fd);
633 Jmsg(jcr, M_FATAL, 0, _("Fseek on attributes file failed: ERR=%s\n"),
638 if (spool_stats.attr_size + size > spool_stats.max_attr_size) {
639 spool_stats.max_attr_size = spool_stats.attr_size + size;
641 spool_stats.attr_size += size;
643 set_jcr_job_status(jcr, JS_AttrDespooling);
644 dir_send_job_status(jcr);
645 Jmsg(jcr, M_INFO, 0, _("Sending spooled attrs to the Director. Despooling %s bytes ...\n"),
646 edit_uint64_with_commas(size, ec1));
647 jcr->dir_bsock->despool(update_attr_spool_size, size);
648 return close_attr_spool_file(jcr, jcr->dir_bsock);
653 close_attr_spool_file(jcr, jcr->dir_bsock);
657 static void make_unique_spool_filename(JCR *jcr, POOLMEM **name, int fd)
659 Mmsg(name, "%s/%s.attr.%s.%d.spool", working_directory, my_name,
664 bool open_attr_spool_file(JCR *jcr, BSOCK *bs)
666 POOLMEM *name = get_pool_memory(PM_MESSAGE);
668 make_unique_spool_filename(jcr, &name, bs->m_fd);
669 bs->m_spool_fd = fopen(name, "w+b");
670 if (!bs->m_spool_fd) {
672 Jmsg(jcr, M_FATAL, 0, _("fopen attr spool file %s failed: ERR=%s\n"), name,
674 free_pool_memory(name);
678 spool_stats.attr_jobs++;
680 free_pool_memory(name);
684 bool close_attr_spool_file(JCR *jcr, BSOCK *bs)
688 if (!bs->m_spool_fd) {
691 name = get_pool_memory(PM_MESSAGE);
693 spool_stats.attr_jobs--;
694 spool_stats.total_attr_jobs++;
696 make_unique_spool_filename(jcr, &name, bs->m_fd);
697 fclose(bs->m_spool_fd);
699 free_pool_memory(name);
700 bs->m_spool_fd = NULL;
701 bs->clear_spooling();