]> git.sur5r.net Git - bacula/bacula/blobdiff - bacula/src/stored/spool.c
update version
[bacula/bacula] / bacula / src / stored / spool.c
index 401cbe061288fd9f550b3c7c6ae6ef7f8104d48e..bd8639dd7d3976cf123d14309e3fcf98e27168c2 100644 (file)
@@ -1,21 +1,14 @@
-/*
- *  Spooling code
- *
- *      Kern Sibbald, March 2004
- *
- *  Version $Id$
- */
 /*
    Bacula® - The Network Backup Solution
 
-   Copyright (C) 2004-2006 Free Software Foundation Europe e.V.
+   Copyright (C) 2004-2010 Free Software Foundation Europe e.V.
 
    The main author of Bacula is Kern Sibbald, with contributions from
    many others, a complete list can be found in the file AUTHORS.
    This program is Free Software; you can redistribute it and/or
    modify it under the terms of version two of the GNU General Public
-   License as published by the Free Software Foundation plus additions
-   that are listed in the file LICENSE.
+   License as published by the Free Software Foundation and included
+   in the file LICENSE.
 
    This program is distributed in the hope that it will be useful, but
    WITHOUT ANY WARRANTY; without even the implied warranty of
    Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA
    02110-1301, USA.
 
-   Bacula® is a registered trademark of John Walker.
+   Bacula® is a registered trademark of Kern Sibbald.
    The licensor of Bacula is the Free Software Foundation Europe
    (FSFE), Fiduciary Program, Sumatrastrasse 25, 8006 Zürich,
    Switzerland, email:ftf@fsfeurope.org.
 */
+/*
+ *  Spooling code
+ *
+ *      Kern Sibbald, March 2004
+ *
+ */
 
 #include "bacula.h"
 #include "stored.h"
@@ -77,10 +76,11 @@ enum {
 
 void list_spool_stats(void sendit(const char *msg, int len, void *sarg), void *arg)
 {
-   char *msg, ed1[30], ed2[30];
+   char ed1[30], ed2[30];
+   POOL_MEM msg(PM_MESSAGE);
    int len;
 
-   msg = (char *)get_pool_memory(PM_MESSAGE);
+   len = Mmsg(msg, _("Spooling statistics:\n"));
 
    if (spool_stats.data_jobs || spool_stats.max_data_size) {
       len = Mmsg(msg, _("Data spooling: %u active jobs, %s bytes; %u total jobs, %s max bytes/job.\n"),
@@ -88,7 +88,7 @@ void list_spool_stats(void sendit(const char *msg, int len, void *sarg), void *a
          spool_stats.total_data_jobs,
          edit_uint64_with_commas(spool_stats.max_data_size, ed2));
 
-      sendit(msg, len, arg);
+      sendit(msg.c_str(), len, arg);
    }
    if (spool_stats.attr_jobs || spool_stats.max_attr_size) {
       len = Mmsg(msg, _("Attr spooling: %u active jobs, %s bytes; %u total jobs, %s max bytes.\n"),
@@ -96,10 +96,8 @@ void list_spool_stats(void sendit(const char *msg, int len, void *sarg), void *a
          spool_stats.total_attr_jobs,
          edit_uint64_with_commas(spool_stats.max_attr_size, ed2));
    
-      sendit(msg, len, arg);
+      sendit(msg.c_str(), len, arg);
    }
-
-   free_pool_memory(msg);
 }
 
 bool begin_data_spool(DCR *dcr)
@@ -154,8 +152,8 @@ static void make_unique_data_spool_filename(DCR *dcr, POOLMEM **name)
    } else {
       dir = working_directory;
    }
-   Mmsg(name, "%s/%s.data.%s.%s.spool", dir, my_name, dcr->jcr->Job, 
-        dcr->device->hdr.name);
+   Mmsg(name, "%s/%s.data.%u.%s.%s.spool", dir, my_name, dcr->jcr->JobId,
+        dcr->jcr->Job, dcr->device->hdr.name);
 }
 
 
@@ -171,7 +169,7 @@ static bool open_data_spool_file(DCR *dcr)
    } else {
       berrno be;
       Jmsg(dcr->jcr, M_FATAL, 0, _("Open data spool file %s failed: ERR=%s\n"), name,
-           be.strerror());
+           be.bstrerror());
       free_pool_memory(name);
       return false;
    }
@@ -207,6 +205,10 @@ static bool close_data_spool_file(DCR *dcr)
 
 static const char *spool_name = "*spool*";
 
+/*
+ * NB! This routine locks the device, but if committing will
+ *     not unlock it. If not committing, it will be unlocked.
+ */
 static bool despool_data(DCR *dcr, bool commit)
 {
    DEVICE *rdev;
@@ -218,7 +220,12 @@ static bool despool_data(DCR *dcr, bool commit)
    char ec1[50];
 
    Dmsg0(100, "Despooling data\n");
-   /* Commit means that the job is done, so we commit, otherwise, we
+   if (jcr->dcr->job_spool_size == 0) {
+      Jmsg(jcr, M_WARNING, 0, _("Despooling zero bytes. Your disk is probably FULL!\n"));
+   }
+
+   /*
+    * Commit means that the job is done, so we commit, otherwise, we
     *  are despooling because of user spool size max or some error  
     *  (e.g. filesystem full).
     */
@@ -226,16 +233,24 @@ static bool despool_data(DCR *dcr, bool commit)
       Jmsg(jcr, M_INFO, 0, _("Committing spooled data to Volume \"%s\". Despooling %s bytes ...\n"),
          jcr->dcr->VolumeName,
          edit_uint64_with_commas(jcr->dcr->job_spool_size, ec1));
+      set_jcr_job_status(jcr, JS_DataCommitting);
    } else {
       Jmsg(jcr, M_INFO, 0, _("Writing spooled data to Volume. Despooling %s bytes ...\n"),
          edit_uint64_with_commas(jcr->dcr->job_spool_size, ec1));
+      set_jcr_job_status(jcr, JS_DataDespooling);
    }
+   set_jcr_job_status(jcr, JS_DataDespooling);
+   dir_send_job_status(jcr);
    dcr->despool_wait = true;
    dcr->spooling = false;
-   lock_device(dcr->dev);
+   /*
+    * We work with device blocked, but not locked so that
+    *  other threads -- e.g. reservations can lock the device
+    *  structure.
+    */
+   dcr->dblock(BST_DESPOOLING);
    dcr->despool_wait = false;
    dcr->despooling = true;
-   dcr->dev_locked = true;
 
    /*
     * This is really quite kludgy and should be fixed some time.
@@ -251,17 +266,22 @@ static bool despool_data(DCR *dcr, bool commit)
    rdev->max_block_size = dcr->dev->max_block_size;
    rdev->min_block_size = dcr->dev->min_block_size;
    rdev->device = dcr->dev->device;
-   rdcr = new_dcr(NULL, rdev);
+   rdcr = new_dcr(jcr, NULL, rdev);
    rdcr->spool_fd = dcr->spool_fd;
-   rdcr->jcr = jcr;                   /* set a valid jcr */
    block = dcr->block;                /* save block */
    dcr->block = rdcr->block;          /* make read and write block the same */
 
    Dmsg1(800, "read/write block size = %d\n", block->buf_len);
    lseek(rdcr->spool_fd, 0, SEEK_SET); /* rewind */
 
+#if defined(HAVE_POSIX_FADVISE) && defined(POSIX_FADV_WILLNEED)
+   posix_fadvise(rdcr->spool_fd, 0, 0, POSIX_FADV_WILLNEED);
+#endif
+
    /* Add run time, to get current wait time */
-   time_t despool_start = time(NULL) - jcr->run_time;
+   int32_t despool_start = time(NULL) - jcr->run_time;
+
+   set_new_file_parameters(dcr);
 
    for ( ; ok; ) {
       if (job_canceled(jcr)) {
@@ -279,18 +299,31 @@ static bool despool_data(DCR *dcr, bool commit)
       if (!ok) {
          Jmsg2(jcr, M_FATAL, 0, _("Fatal append error on device %s: ERR=%s\n"),
                dcr->dev->print_name(), dcr->dev->bstrerror());
+         Dmsg2(000, "Fatal append error on device %s: ERR=%s\n",
+               dcr->dev->print_name(), dcr->dev->bstrerror());
       }
       Dmsg3(800, "Write block ok=%d FI=%d LI=%d\n", ok, block->FirstIndex, block->LastIndex);
    }
 
-   /* Subtracting run_time give us elapsed time - wait_time since we started despooling */
-   time_t despool_elapsed = time(NULL) - despool_start - jcr->run_time;
+   if (!dir_create_jobmedia_record(dcr)) {
+      Jmsg2(jcr, M_FATAL, 0, _("Could not create JobMedia record for Volume=\"%s\" Job=%s\n"),
+         dcr->getVolCatName(), jcr->Job);
+   }
+   /* Set new file/block parameters for current dcr */
+   set_new_file_parameters(dcr);
+
+   /*
+    * Subtracting run_time give us elapsed time - wait_time since 
+    * we started despooling. Note, don't use time_t as it is 32 or 64
+    * bits depending on the OS and doesn't edit with %d
+    */
+   int32_t despool_elapsed = time(NULL) - despool_start - jcr->run_time;
 
    if (despool_elapsed <= 0) {
       despool_elapsed = 1;
    }
 
-   Jmsg(dcr->jcr, M_INFO, 0, _("Despooling elapsed time = %02d:%02d:%02d, Transfer rate = %s bytes/second\n"),
+   Jmsg(dcr->jcr, M_INFO, 0, _("Despooling elapsed time = %02d:%02d:%02d, Transfer rate = %s Bytes/second\n"),
          despool_elapsed / 3600, despool_elapsed % 3600 / 60, despool_elapsed % 60,
          edit_uint64_with_suffix(jcr->dcr->job_spool_size / despool_elapsed, ec1));
 
@@ -300,9 +333,8 @@ static bool despool_data(DCR *dcr, bool commit)
    if (ftruncate(rdcr->spool_fd, 0) != 0) {
       berrno be;
       Jmsg(dcr->jcr, M_ERROR, 0, _("Ftruncate spool file failed: ERR=%s\n"),
-         be.strerror());
-      Pmsg1(000, _("Bad return from ftruncate. ERR=%s\n"), be.strerror());
-      ok = false;
+         be.bstrerror());
+      /* Note, try continuing despite ftruncate problem */
    }
 
    P(mutex);
@@ -320,15 +352,20 @@ static bool despool_data(DCR *dcr, bool commit)
    free_pool_memory(rdev->errmsg);
    /* Be careful to NULL the jcr and free rdev after free_dcr() */
    rdcr->jcr = NULL;
+   rdcr->dev = NULL;
    free_dcr(rdcr);
    free(rdev);
    dcr->spooling = true;           /* turn on spooling again */
    dcr->despooling = false;
-   /* If doing a commit, leave the device locked -- unlocked in release_device() */
+   /*
+    * Note, if committing we leave the device blocked. It will be removed in
+    *  release_device();
+    */
    if (!commit) {
-      dcr->dev_locked = false;
-      unlock_device(dcr->dev);
+      dcr->dev->dunblock();
    }
+   set_jcr_job_status(jcr, JS_Running);
+   dir_send_job_status(jcr);
    return ok;
 }
 
@@ -355,7 +392,7 @@ static int read_block_from_spool_file(DCR *dcr)
       if (stat == -1) {
          berrno be;
          Jmsg(dcr->jcr, M_FATAL, 0, _("Spool header read error. ERR=%s\n"),
-              be.strerror());
+              be.bstrerror());
       } else {
          Pmsg2(000, _("Spool read error. Wanted %u bytes, got %d\n"), rlen, stat);
          Jmsg2(dcr->jcr, M_FATAL, 0, _("Spool header read error. Wanted %u bytes, got %d\n"), rlen, stat);
@@ -397,6 +434,9 @@ bool write_block_to_spool_file(DCR *dcr)
    bool despool = false;
    DEV_BLOCK *block = dcr->block;
 
+   if (job_canceled(dcr->jcr)) {
+      return false;
+   }
    ASSERT(block->binbuf == ((uint32_t) (block->bufp - block->buf)));
    if (block->binbuf <= WRITE_BLKHDR_LENGTH) {  /* Does block have data in it? */
       return true;
@@ -470,21 +510,24 @@ static bool write_spool_header(DCR *dcr)
       if (stat == -1) {
          berrno be;
          Jmsg(dcr->jcr, M_FATAL, 0, _("Error writing header to spool file. ERR=%s\n"),
-              be.strerror());
+              be.bstrerror());
       }
       if (stat != (ssize_t)sizeof(hdr)) {
+         Jmsg(dcr->jcr, M_ERROR, 0, _("Error writing header to spool file."
+              " Disk probably full. Attempting recovery. Wanted to write=%d got=%d\n"),
+              (int)stat, (int)sizeof(hdr));
          /* If we wrote something, truncate it, then despool */
          if (stat != -1) {
 #if defined(HAVE_WIN32)
             boffset_t   pos = _lseeki64(dcr->spool_fd, (__int64)0, SEEK_CUR);
 #else
-            boffset_t   pos = lseek(dcr->spool_fd, (off_t)0, SEEK_CUR);
+            boffset_t   pos = lseek(dcr->spool_fd, 0, SEEK_CUR);
 #endif
             if (ftruncate(dcr->spool_fd, pos - stat) != 0) {
                berrno be;
-               Jmsg(dcr->jcr, M_FATAL, 0, _("Ftruncate spool file failed: ERR=%s\n"),
-                  be.strerror());
-               return false;
+               Jmsg(dcr->jcr, M_ERROR, 0, _("Ftruncate spool file failed: ERR=%s\n"),
+                  be.bstrerror());
+              /* Note, try continuing despite ftruncate problem */
             }
          }
          if (!despool_data(dcr, false)) {
@@ -510,7 +553,7 @@ static bool write_spool_data(DCR *dcr)
       if (stat == -1) {
          berrno be;
          Jmsg(dcr->jcr, M_FATAL, 0, _("Error writing data to spool file. ERR=%s\n"),
-              be.strerror());
+              be.bstrerror());
       }
       if (stat != (ssize_t)block->binbuf) {
          /*
@@ -520,13 +563,13 @@ static bool write_spool_data(DCR *dcr)
 #if defined(HAVE_WIN32)
             boffset_t   pos = _lseeki64(dcr->spool_fd, (__int64)0, SEEK_CUR);
 #else
-            boffset_t   pos = lseek(dcr->spool_fd, (off_t)0, SEEK_CUR);
+            boffset_t   pos = lseek(dcr->spool_fd, 0, SEEK_CUR);
 #endif
             if (ftruncate(dcr->spool_fd, pos - stat - sizeof(spool_hdr)) != 0) {
                berrno be;
-               Jmsg(dcr->jcr, M_FATAL, 0, _("Ftruncate spool file failed: ERR=%s\n"),
-                  be.strerror());
-               return false;
+               Jmsg(dcr->jcr, M_ERROR, 0, _("Ftruncate spool file failed: ERR=%s\n"),
+                  be.bstrerror());
+               /* Note, try continuing despite ftruncate problem */
             }
          }
          if (!despool_data(dcr, false)) {
@@ -548,7 +591,7 @@ static bool write_spool_data(DCR *dcr)
 
 bool are_attributes_spooled(JCR *jcr)
 {
-   return jcr->spool_attributes && jcr->dir_bsock->spool_fd;
+   return jcr->spool_attributes && jcr->dir_bsock->m_spool_fd;
 }
 
 /*
@@ -587,23 +630,58 @@ static void update_attr_spool_size(ssize_t size)
    V(mutex);
 }
 
+static void make_unique_spool_filename(JCR *jcr, POOLMEM **name, int fd)
+{
+   Mmsg(name, "%s/%s.attr.%s.%d.spool", working_directory, my_name,
+      jcr->Job, fd);
+}
+
+/*
+ * Tell Director where to find the attributes spool file 
+ *  Note, if we are not on the same machine, the Director will
+ *  return an error, and the higher level routine will transmit
+ *  the data record by record -- using bsock->despool().
+ */
+static bool blast_attr_spool_file(JCR *jcr, boffset_t size)
+{
+   /* send full spool file name */
+   POOLMEM *name  = get_pool_memory(PM_MESSAGE);
+   make_unique_spool_filename(jcr, &name, jcr->dir_bsock->m_fd);
+   bash_spaces(name);
+   jcr->dir_bsock->fsend("BlastAttr Job=%s File=%s\n", jcr->Job, name);
+   free_pool_memory(name);
+   
+   if (jcr->dir_bsock->recv() <= 0) {
+      Jmsg(jcr, M_FATAL, 0, _("Network error on BlastAttributes.\n"));
+      return false;
+   }
+   
+   if (!bstrcmp(jcr->dir_bsock->msg, "1000 OK BlastAttr\n")) {
+      return false;
+   }
+   return true;
+}
+
 bool commit_attribute_spool(JCR *jcr)
 {
-   off_t size;
+   boffset_t size;
    char ec1[30];
+   char tbuf[100];
 
+   Dmsg1(100, "Commit attributes at %s\n", bstrftimes(tbuf, sizeof(tbuf),
+         (utime_t)time(NULL)));
    if (are_attributes_spooled(jcr)) {
-      if (fseeko(jcr->dir_bsock->spool_fd, 0, SEEK_END) != 0) {
+      if (fseeko(jcr->dir_bsock->m_spool_fd, 0, SEEK_END) != 0) {
          berrno be;
          Jmsg(jcr, M_FATAL, 0, _("Fseek on attributes file failed: ERR=%s\n"),
-              be.strerror());
+              be.bstrerror());
          goto bail_out;
       }
-      size = ftello(jcr->dir_bsock->spool_fd);
+      size = ftello(jcr->dir_bsock->m_spool_fd);
       if (size < 0) {
          berrno be;
          Jmsg(jcr, M_FATAL, 0, _("Fseek on attributes file failed: ERR=%s\n"),
-              be.strerror());
+              be.bstrerror());
          goto bail_out;
       }
       P(mutex);
@@ -612,9 +690,17 @@ bool commit_attribute_spool(JCR *jcr)
       }
       spool_stats.attr_size += size;
       V(mutex);
+      set_jcr_job_status(jcr, JS_AttrDespooling);
+      dir_send_job_status(jcr);
       Jmsg(jcr, M_INFO, 0, _("Sending spooled attrs to the Director. Despooling %s bytes ...\n"),
             edit_uint64_with_commas(size, ec1));
-      bnet_despool_to_bsock(jcr->dir_bsock, update_attr_spool_size, size);
+
+      if (!blast_attr_spool_file(jcr, size)) {
+         /* Can't read spool file from director side,
+          * send content over network.
+          */
+         jcr->dir_bsock->despool(update_attr_spool_size, size);
+      }
       return close_attr_spool_file(jcr, jcr->dir_bsock);
    }
    return true;
@@ -624,23 +710,16 @@ bail_out:
    return false;
 }
 
-static void make_unique_spool_filename(JCR *jcr, POOLMEM **name, int fd)
-{
-   Mmsg(name, "%s/%s.attr.%s.%d.spool", working_directory, my_name,
-      jcr->Job, fd);
-}
-
-
-bool open_attr_spool_file(JCR *jcr, BSOCK *bs)
+static bool open_attr_spool_file(JCR *jcr, BSOCK *bs)
 {
    POOLMEM *name  = get_pool_memory(PM_MESSAGE);
 
-   make_unique_spool_filename(jcr, &name, bs->fd);
-   bs->spool_fd = fopen(name, "w+b");
-   if (!bs->spool_fd) {
+   make_unique_spool_filename(jcr, &name, bs->m_fd);
+   bs->m_spool_fd = fopen(name, "w+b");
+   if (!bs->m_spool_fd) {
       berrno be;
       Jmsg(jcr, M_FATAL, 0, _("fopen attr spool file %s failed: ERR=%s\n"), name,
-           be.strerror());
+           be.bstrerror());
       free_pool_memory(name);
       return false;
    }
@@ -651,11 +730,15 @@ bool open_attr_spool_file(JCR *jcr, BSOCK *bs)
    return true;
 }
 
-bool close_attr_spool_file(JCR *jcr, BSOCK *bs)
+static bool close_attr_spool_file(JCR *jcr, BSOCK *bs)
 {
    POOLMEM *name;
 
-   if (!bs->spool_fd) {
+   char tbuf[100];
+
+   Dmsg1(100, "Close attr spool file at %s\n", bstrftimes(tbuf, sizeof(tbuf),
+         (utime_t)time(NULL)));
+   if (!bs->m_spool_fd) {
       return true;
    }
    name = get_pool_memory(PM_MESSAGE);
@@ -663,11 +746,11 @@ bool close_attr_spool_file(JCR *jcr, BSOCK *bs)
    spool_stats.attr_jobs--;
    spool_stats.total_attr_jobs++;
    V(mutex);
-   make_unique_spool_filename(jcr, &name, bs->fd);
-   fclose(bs->spool_fd);
+   make_unique_spool_filename(jcr, &name, bs->m_fd);
+   fclose(bs->m_spool_fd);
    unlink(name);
    free_pool_memory(name);
-   bs->spool_fd = NULL;
-   bs->spool = false;
+   bs->m_spool_fd = NULL;
+   bs->clear_spooling();
    return true;
 }