git.sur5r.net Git - bacula/bacula/blob - bacula/patches/testing/blastattributes.patch

   1 Index: src/dird/catreq.c
   2 ===================================================================
   3 --- src/dird/catreq.c   (révision 8270)
   4 +++ src/dird/catreq.c   (copie de travail)
   5 @@ -356,14 +356,7 @@
   6     return;
   7  }
   8
   9 -/*
  10 - * Update File Attributes in the catalog with data
  11 - *  sent by the Storage daemon.  Note, we receive the whole
  12 - *  attribute record, but we select out only the stat packet,
  13 - *  VolSessionId, VolSessionTime, FileIndex, file type, and
  14 - *  file name to store in the catalog.
  15 - */
  16 -void catalog_update(JCR *jcr, BSOCK *bs)
  17 +static void insert_attribute(JCR *jcr, char *msg, int32_t msglen)
  18  {
  19     unser_declare;
  20     uint32_t VolSessionId, VolSessionTime;
  21 @@ -375,20 +368,7 @@
  22     int len;
  23     char *fname, *attr;
  24     ATTR_DBR *ar = NULL;
  25 -   POOLMEM *omsg;
  26
  27 -   if (job_canceled(jcr) || !jcr->pool->catalog_files) {
  28 -      goto bail_out;                  /* user disabled cataloging */
  29 -   }
  30 -   if (!jcr->db) {
  31 -      omsg = get_memory(bs->msglen+1);
  32 -      pm_strcpy(omsg, bs->msg);
  33 -      bs->fsend(_("1994 Invalid Catalog Update: %s"), omsg);
  34 -      Jmsg1(jcr, M_FATAL, 0, _("Invalid Catalog Update; DB not open: %s"), omsg);
  35 -      free_memory(omsg);
  36 -      goto bail_out;
  37 -   }
  38 -
  39     /* Start transaction allocates jcr->attr and jcr->ar if needed */
  40     db_start_transaction(jcr, jcr->db);     /* start transaction if not already open */
  41     ar = jcr->ar;
  42 @@ -397,7 +377,7 @@
  43      *  there may be a cached attr so we cannot yet write into
  44      *  jcr->attr or jcr->ar
  45      */
  46 -   p = bs->msg;
  47 +   p = msg;
  48     skip_nonspaces(&p);                /* UpdCat */
  49     skip_spaces(&p);
  50     skip_nonspaces(&p);                /* Job=nnn */
  51 @@ -412,7 +392,7 @@
  52     unser_uint32(data_len);
  53     p += unser_length(p);
  54
  55 -   Dmsg1(400, "UpdCat msg=%s\n", bs->msg);
  56 +   Dmsg1(400, "UpdCat msg=%s\n", msg);
  57     Dmsg5(400, "UpdCat VolSessId=%d VolSessT=%d FI=%d Strm=%d data_len=%d\n",
  58        VolSessionId, VolSessionTime, FileIndex, Stream, data_len);
  59
  60 @@ -424,9 +404,9 @@
  61           }
  62        }
  63        /* Any cached attr is flushed so we can reuse jcr->attr and jcr->ar */
  64 -      jcr->attr = check_pool_memory_size(jcr->attr, bs->msglen);
  65 -      memcpy(jcr->attr, bs->msg, bs->msglen);
  66 -      p = jcr->attr - bs->msg + p;    /* point p into jcr->attr */
  67 +      jcr->attr = check_pool_memory_size(jcr->attr, msglen);
  68 +      memcpy(jcr->attr, msg, msglen);
  69 +      p = jcr->attr - msg + p;    /* point p into jcr->attr */
  70        skip_nonspaces(&p);             /* skip FileIndex */
  71        skip_spaces(&p);
  72        filetype = str_to_int32(p);     /* TODO: choose between unserialize and str_to_int32 */
  73 @@ -510,8 +490,104 @@
  74           }
  75        }
  76     }
  77 +}
  78 +
  79 +/*
  80 + * Update File Attributes in the catalog with data
  81 + *  sent by the Storage daemon.  Note, we receive the whole
  82 + *  attribute record, but we select out only the stat packet,
  83 + *  VolSessionId, VolSessionTime, FileIndex, file type, and
  84 + *  file name to store in the catalog.
  85 + */
  86 +void catalog_update(JCR *jcr, BSOCK *bs)
  87 +{
  88 +   POOLMEM *omsg;
  89 +
  90 +   if (job_canceled(jcr) || !jcr->pool->catalog_files) {
  91 +      goto bail_out;                  /* user disabled cataloging */
  92 +   }
  93 +   if (!jcr->db) {
  94 +      omsg = get_memory(bs->msglen+1);
  95 +      pm_strcpy(omsg, bs->msg);
  96 +      bs->fsend(_("1994 Invalid Catalog Update: %s"), omsg);
  97 +      Jmsg1(jcr, M_FATAL, 0, _("Invalid Catalog Update; DB not open: %s"), omsg);
  98 +      free_memory(omsg);
  99 +      goto bail_out;
 100 +
 101 +   }
 102 +
 103 +   insert_attribute(jcr, bs->msg, bs->msglen);
 104 +
 105  bail_out:
 106     if (job_canceled(jcr)) {
 107        cancel_storage_daemon_job(jcr);
 108     }
 109  }
 110 +
 111 +bool despool_attributes_from_file(JCR *jcr, const char *file)
 112 +{
 113 +   bool ret=false;
 114 +   int32_t pktsiz;
 115 +   size_t nbytes;
 116 +   ssize_t last = 0, size = 0;
 117 +   int count = 0;
 118 +   int32_t msglen;                    /* message length */
 119 +   POOLMEM *msg = get_pool_memory(PM_MESSAGE);
 120 +   FILE *spool_fd;
 121 +
 122 +   Dmsg0(0, "Begin despool_attributes_from_file\n");
 123 +
 124 +   spool_fd = fopen(file, "rb");
 125 +   if (!spool_fd) {
 126 +      Dmsg0(0, "cancel despool_attributes_from_file\n");
 127 +      /* send an error message */
 128 +      goto bail_out;
 129 +   }
 130 +#if defined(HAVE_POSIX_FADVISE) && defined(POSIX_FADV_WILLNEED)
 131 +   posix_fadvise(fileno(spool_fd), 0, 0, POSIX_FADV_WILLNEED);
 132 +#endif
 133 +
 134 +   while (fread((char *)&pktsiz, 1, sizeof(int32_t), spool_fd) ==
 135 +          sizeof(int32_t)) {
 136 +      size += sizeof(int32_t);
 137 +      msglen = ntohl(pktsiz);
 138 +      if (msglen > 0) {
 139 +         if (msglen > (int32_t) sizeof_pool_memory(msg)) {
 140 +            msg = realloc_pool_memory(msg, msglen + 1);
 141 +         }
 142 +         nbytes = fread(msg, 1, msglen, spool_fd);
 143 +         if (nbytes != (size_t) msglen) {
 144 +            berrno be;
 145 +            Dmsg2(400, "nbytes=%d msglen=%d\n", nbytes, msglen);
 146 +            Qmsg1(jcr, M_FATAL, 0, _("fread attr spool error. ERR=%s\n"),
 147 +                  be.bstrerror());
 148 +            goto bail_out;
 149 +         }
 150 +         size += nbytes;
 151 +         if ((++count & 0x3F) == 0) {
 152 +            last = size;
 153 +         }
 154 +      }
 155 +      insert_attribute(jcr, msg, msglen);
 156 +   }
 157 +   if (ferror(spool_fd)) {
 158 +      berrno be;
 159 +      Qmsg1(jcr, M_FATAL, 0, _("fread attr spool error. ERR=%s\n"),
 160 +            be.bstrerror());
 161 +      goto bail_out;
 162 +   }
 163 +   ret = true;
 164 +
 165 +bail_out:
 166 +   if (spool_fd) {
 167 +      fclose(spool_fd);
 168 +   }
 169 +
 170 +   if (job_canceled(jcr)) {
 171 +      cancel_storage_daemon_job(jcr);
 172 +   }
 173 +
 174 +   free_pool_memory(msg);
 175 +   Dmsg1(0, "End despool_attributes_from_file ret=%i\n", ret);
 176 +   return ret;
 177 +}
 178 Index: src/dird/getmsg.c
 179 ===================================================================
 180 --- src/dird/getmsg.c   (révision 8270)
 181 +++ src/dird/getmsg.c   (copie de travail)
 182 @@ -135,7 +135,7 @@
 183
 184     for (;;) {
 185        n = bs->recv();
 186 -      Dmsg2(300, "bget_dirmsg %d: %s\n", n, bs->msg);
 187 +      Dmsg2(100, "bget_dirmsg %d: %s\n", n, bs->msg);
 188
 189        if (is_bnet_stop(bs)) {
 190           return n;                    /* error or terminate */
 191 @@ -236,6 +236,22 @@
 192           catalog_update(jcr, bs);
 193           continue;
 194        }
 195 +      if (bs->msg[0] == 'B') {        /* SD sending file spool attributes */
 196 +         Dmsg2(0, "Blast attributes jcr 0x%x: %s", jcr, bs->msg);
 197 +         char filename[MAX_NAME_LENGTH];
 198 +         if (sscanf(bs->msg, "BlastAttr Job=%127s File=%127s",
 199 +                    Job, filename) != 2) {
 200 +            Jmsg1(jcr, M_ERROR, 0, _("Malformed message: %s\n"), bs->msg);
 201 +            continue;
 202 +         }
 203 +         unbash_spaces(filename);
 204 +         if (despool_attributes_from_file(jcr, filename)) {
 205 +            bs->fsend("1000 OK BlastAttr\n");
 206 +         } else {
 207 +            bs->fsend("1990 ERROR BlastAttr\n");
 208 +         }
 209 +         continue;
 210 +      }
 211        if (bs->msg[0] == 'M') {        /* Mount request */
 212           Dmsg1(900, "Mount req: %s", bs->msg);
 213           mount_request(jcr, bs, msg);
 214 Index: src/dird/protos.h
 215 ===================================================================
 216 --- src/dird/protos.h   (révision 8270)
 217 +++ src/dird/protos.h   (copie de travail)
 218 @@ -80,6 +80,7 @@
 219  /* catreq.c */
 220  extern void catalog_request(JCR *jcr, BSOCK *bs);
 221  extern void catalog_update(JCR *jcr, BSOCK *bs);
 222 +extern bool despool_attributes_from_file(JCR *jcr, const char *file);
 223
 224  /* dird_conf.c */
 225  extern const char *level_to_str(int level);
 226 Index: src/stored/spool.c
 227 ===================================================================
 228 --- src/stored/spool.c  (révision 8270)
 229 +++ src/stored/spool.c  (copie de travail)
 230 @@ -622,13 +622,40 @@
 231     V(mutex);
 232  }
 233
 234 +static void make_unique_spool_filename(JCR *jcr, POOLMEM **name, int fd)
 235 +{
 236 +   Mmsg(name, "%s/%s.attr.%s.%d.spool", working_directory, my_name,
 237 +      jcr->Job, fd);
 238 +}
 239 +
 240 +static bool blast_attr_spool_file(JCR *jcr, boffset_t size)
 241 +{
 242 +   /* send full spool file name */
 243 +   POOLMEM *name  = get_pool_memory(PM_MESSAGE);
 244 +   make_unique_spool_filename(jcr, &name, jcr->dir_bsock->m_fd);
 245 +   bash_spaces(name);
 246 +   jcr->dir_bsock->fsend("BlastAttr Job=%s File=%s\n",
 247 +                         jcr->Job, name);
 248 +   free_pool_memory(name);
 249 +
 250 +   if (jcr->dir_bsock->recv() <= 0) {
 251 +      Jmsg(jcr, M_FATAL, 0, _("Network error on BlastAttributes.\n"));
 252 +      return false;
 253 +   }
 254 +
 255 +   if (!bstrcmp(jcr->dir_bsock->msg, "1000 OK BlastAttr\n")) {
 256 +      return false;
 257 +   }
 258 +   return true;
 259 +}
 260 +
 261  bool commit_attribute_spool(JCR *jcr)
 262  {
 263     boffset_t size;
 264     char ec1[30];
 265     char tbuf[100];
 266
 267 -   Dmsg1(100, "Commit attributes at %s\n", bstrftimes(tbuf, sizeof(tbuf),
 268 +   Dmsg1(0, "Commit attributes at %s\n", bstrftimes(tbuf, sizeof(tbuf),
 269           ( utime_t)time(NULL)));
 270     if (are_attributes_spooled(jcr)) {
 271        if (fseeko(jcr->dir_bsock->m_spool_fd, 0, SEEK_END) != 0) {
 272 @@ -654,7 +681,13 @@
 273        dir_send_job_status(jcr);
 274        Jmsg(jcr, M_INFO, 0, _("Sending spooled attrs to the Director. Despooling %s bytes ...\n"),
 275              edit_uint64_with_commas(size, ec1));
 276 -      jcr->dir_bsock->despool(update_attr_spool_size, size);
 277 +
 278 +      if (!blast_attr_spool_file(jcr, size)) {
 279 +         /* Can't read spool file from director side,
 280 +          * send content over network.
 281 +          */
 282 +         jcr->dir_bsock->despool(update_attr_spool_size, size);
 283 +      }
 284        return close_attr_spool_file(jcr, jcr->dir_bsock);
 285     }
 286     return true;
 287 @@ -664,13 +697,6 @@
 288     return false;
 289  }
 290
 291 -static void make_unique_spool_filename(JCR *jcr, POOLMEM **name, int fd)
 292 -{
 293 -   Mmsg(name, "%s/%s.attr.%s.%d.spool", working_directory, my_name,
 294 -      jcr->Job, fd);
 295 -}
 296 -
 297 -
 298  bool open_attr_spool_file(JCR *jcr, BSOCK *bs)
 299  {
 300     POOLMEM *name  = get_pool_memory(PM_MESSAGE);