Backport messages.c from Branch-5.0

diff --git a/bacula/src/jcr.h b/bacula/src/jcr.h
index 26da03c3a793c399a870bceaf6ac9502119dafe7..49402caf1d361156579b3e7134e6bdf339e33f01 100644 (file)
--- a/bacula/src/jcr.h
+++ b/bacula/src/jcr.h
@@ -202,8 +202,9 @@ public:
    BSOCK *file_bsock;                 /* File daemon connection socket */
    JCR_free_HANDLER *daemon_free_jcr; /* Local free routine */
    dlist *msg_queue;                  /* Queued messages */
+   pthread_mutex_t msg_queue_mutex;   /* message queue mutex */
+   bool dequeuing_msgs;               /* Set when dequeuing messages */
    alist job_end_push;                /* Job end pushed calls */
-   bool dequeuing;                    /* dequeuing messages */
    POOLMEM *VolumeName;               /* Volume name desired -- pool_memory */
    POOLMEM *errmsg;                   /* edited error message */
    char Job[MAX_NAME_LENGTH];         /* Unique name of this Job */

diff --git a/bacula/src/lib/jcr.c b/bacula/src/lib/jcr.c
index 85b79c5fa0d3aca8328af51c55ce149d43c78f59..4596c175b4c9c8734b2b6d1186714567e3f74e63 100644 (file)
--- a/bacula/src/lib/jcr.c
+++ b/bacula/src/lib/jcr.c
@@ -355,6 +355,11 @@ JCR *new_jcr(int size, JCR_free_HANDLER *daemon_free_jcr)
    memset(jcr, 0, size);
    jcr->my_thread_id = pthread_self();
    jcr->msg_queue = New(dlist(item, &item->link));
+   if ((status = pthread_mutex_init(&jcr->msg_queue_mutex, NULL)) != 0) {
+      berrno be;
+      Jmsg(NULL, M_ABORT, 0, _("Could not init msg_queue mutex. ERR=%s\n"),
+         be.bstrerror(status));
+   }
    jcr->job_end_push.init(1, false);
    jcr->sched_time = time(NULL);
    jcr->daemon_free_jcr = daemon_free_jcr;    /* plug daemon free routine */
@@ -421,6 +426,7 @@ static void free_common_jcr(JCR *jcr)
    if (jcr->msg_queue) {
       delete jcr->msg_queue;
       jcr->msg_queue = NULL;
+      pthread_mutex_destroy(&jcr->msg_queue_mutex);
    }
    close_msg(jcr);                    /* close messages for this job */
 
@@ -1085,7 +1091,6 @@ void _dbg_print_jcr(FILE *fp)
       bstrftime(buf4, sizeof(buf4), jcr->wait_time);
       fprintf(fp, "\tsched_time=%s start_time=%s\n\tend_time=%s wait_time=%s\n",
               buf1, buf2, buf3, buf4);
-      fprintf(fp, "\tdequeing=%i\n", jcr->dequeuing);
       fprintf(fp, "\tdb=%p db_batch=%p batch_started=%i\n", 
               jcr->db, jcr->db_batch, jcr->batch_started);
       

diff --git a/bacula/src/lib/message.c b/bacula/src/lib/message.c
index 9c18e5ad9cd0c1b46fdf4249403353534e9d606c..6d951e2bbfd20fec13cf0bd1f40af084ddda94bc 100644 (file)
--- a/bacula/src/lib/message.c
+++ b/bacula/src/lib/message.c
@@ -1,7 +1,7 @@
 /*
    Bacula® - The Network Backup Solution
 
-   Copyright (C) 2000-2009 Free Software Foundation Europe e.V.
+   Copyright (C) 2000-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.
@@ -28,13 +28,15 @@
 /*
  * Bacula message handling routines
  *
- *   Kern Sibbald, April 2000
+ * NOTE: don't use any Jmsg or Qmsg calls within this file,
+ *   except in q_msg or j_msg (setup routines), 
+ *   otherwise you may get into recursive calls if there are
+ *   errors, and that can lead to looping or deadlocks.
  *
- *   Version $Id$
+ *   Kern Sibbald, April 2000
  *
  */
 
-
 #include "bacula.h"
 #include "jcr.h"
 
@@ -51,6 +53,7 @@ const char *working_directory = NULL;       /* working directory path stored her
 int verbose = 0;                      /* increase User messages */
 int debug_level = 0;                  /* debug level */
 bool dbg_timestamp = false;           /* print timestamp in debug output */
+bool prt_kaboom = false;              /* Print kaboom output */
 utime_t daemon_start_time = 0;        /* Daemon start time */
 const char *version = VERSION " (" BDATE ")";
 char my_name[30];                     /* daemon name is stored here */
@@ -86,6 +89,43 @@ const char *host_os = HOST_OS;
 const char *distname = DISTNAME;
 const char *distver = DISTVER;
 
+/*
+ * Handle message delivery errors
+ */
+static void delivery_error(const char *fmt,...)
+{
+   va_list   arg_ptr;
+   int i, len, maxlen;
+   POOLMEM *pool_buf;
+   char dt[MAX_TIME_LENGTH];
+   int dtlen;
+
+   pool_buf = get_pool_memory(PM_EMSG);
+
+   bstrftime_ny(dt, sizeof(dt), time(NULL));
+   dtlen = strlen(dt);
+   dt[dtlen++] = ' ';
+   dt[dtlen] = 0;
+
+   i = Mmsg(pool_buf, "%s Message delivery ERROR: ", dt);
+
+   for (;;) {
+      maxlen = sizeof_pool_memory(pool_buf) - i - 1;
+      va_start(arg_ptr, fmt);
+      len = bvsnprintf(pool_buf+i, maxlen, fmt, arg_ptr);
+      va_end(arg_ptr);
+      if (len < 0 || len >= (maxlen-5)) {
+         pool_buf = realloc_pool_memory(pool_buf, maxlen + i + maxlen/2);
+         continue;
+      }
+      break;
+   }
+
+   fputs(pool_buf, stdout);  /* print this here to INSURE that it is printed */
+   fflush(stdout);
+   syslog(LOG_DAEMON|LOG_ERR, "%s", pool_buf);
+   free_memory(pool_buf);
+}                 
 
 void register_message_callback(void msg_callback(int type, char *msg))
 {
@@ -399,7 +439,7 @@ static BPIPE *open_mail_pipe(JCR *jcr, POOLMEM *&cmd, DEST *d)
       }
    } else {
       berrno be;
-      Jmsg(jcr, M_ERROR, 0, _("open mail pipe %s failed: ERR=%s\n"),
+      delivery_error(_("open mail pipe %s failed: ERR=%s\n"),
          cmd, be.bstrerror());
    }
    return bpipe;
@@ -482,7 +522,7 @@ void close_msg(JCR *jcr)
             if (msgs != daemon_msgs) {
                /* read what mail prog returned -- should be nothing */
                while (fgets(line, len, bpipe->rfd)) {
-                  Jmsg1(jcr, M_INFO, 0, _("Mail prog: %s"), line);
+                  delivery_error(_("Mail prog: %s"), line);
                }
             }
 
@@ -491,9 +531,9 @@ void close_msg(JCR *jcr)
                berrno be;
                be.set_errno(stat);
                Dmsg1(850, "Calling emsg. CMD=%s\n", cmd);
-               Jmsg2(jcr, M_ERROR, 0, _("Mail program terminated in error.\n"
-                                        "CMD=%s\n"
-                                        "ERR=%s\n"), cmd, be.bstrerror());
+               delivery_error(_("Mail program terminated in error.\n"
+                                 "CMD=%s\n"
+                                 "ERR=%s\n"), cmd, be.bstrerror());
             }
             free_memory(line);
 rem_temp_file:
@@ -588,9 +628,7 @@ static bool open_dest_file(JCR *jcr, DEST *d, const char *mode)
    d->fd = fopen(d->where, mode);
    if (!d->fd) {
       berrno be;
-      d->fd = stdout;
-      Qmsg2(jcr, M_ERROR, 0, _("fopen %s failed: ERR=%s\n"), d->where, be.bstrerror());
-      d->fd = NULL;
+      delivery_error(_("fopen %s failed: ERR=%s\n"), d->where, be.bstrerror());
       return false;
    }
    return true;
@@ -623,6 +661,7 @@ void dispatch_message(JCR *jcr, int type, utime_t mtime, char *msg)
     if (mtime == 1) {
        *dt = 0;
        dtlen = 0;
+       mtime = time(NULL);      /* get time for SQL log */
     } else {
        bstrftime_ny(dt, sizeof(dt), mtime);
        dtlen = strlen(dt);
@@ -636,10 +675,14 @@ void dispatch_message(JCR *jcr, int type, utime_t mtime, char *msg)
        return;
     }
 
+    /* For serious errors make sure message is printed or logged */
     if (type == M_ABORT || type == M_ERROR_TERM) {
        fputs(dt, stdout);
-       fputs(msg, stdout);         /* print this here to INSURE that it is printed */
+       fputs(msg, stdout);
        fflush(stdout);
+       if (type == M_ABORT) {
+          syslog(LOG_DAEMON|LOG_ERR, "%s", msg);
+       }
     }
 
 
@@ -724,7 +767,7 @@ void dispatch_message(JCR *jcr, int type, utime_t mtime, char *msg)
                    if (stat != 0) {
                       berrno be;
                       be.set_errno(stat);
-                      Qmsg2(jcr, M_ERROR, 0, _("Operator mail program terminated in error.\n"
+                      delivery_error(_("Msg delivery error: Operator mail program terminated in error.\n"
                             "CMD=%s\n"
                             "ERR=%s\n"), mcmd, be.bstrerror());
                    }
@@ -742,10 +785,8 @@ void dispatch_message(JCR *jcr, int type, utime_t mtime, char *msg)
                    d->fd = fopen(name, "w+b");
                    if (!d->fd) {
                       berrno be;
-                      d->fd = stdout;
-                      Qmsg2(jcr, M_ERROR, 0, _("fopen %s failed: ERR=%s\n"), name,
+                      delivery_error(_("Msg delivery error: fopen %s failed: ERR=%s\n"), name,
                             be.bstrerror());
-                      d->fd = NULL;
                       free_pool_memory(name);
                       V(fides_mutex);
                       break;
@@ -841,6 +882,34 @@ get_basename(const char *pathname)
 #endif
 }
 
+/*
+ * print or write output to trace file 
+ */
+static void pt_out(char *buf)
+{
+    /*
+     * Used the "trace on" command in the console to turn on
+     *  output to the trace file.  "trace off" will close the file.
+     */
+    if (trace) {
+       if (!trace_fd) {
+          char fn[200];
+          bsnprintf(fn, sizeof(fn), "%s/%s.trace", working_directory ? working_directory : "./", my_name);
+          trace_fd = fopen(fn, "a+b");
+       }
+       if (trace_fd) {
+          fputs(buf, trace_fd);
+          fflush(trace_fd);
+       } else {
+          /* Some problem, turn off tracing */
+          trace = false;
+       }
+    } else {   /* not tracing */
+       fputs(buf, stdout);
+       fflush(stdout);
+    }
+}
+
 /*********************************************************************
  *
  *  This subroutine prints a debug message if the level number
@@ -889,27 +958,7 @@ d_msg(const char *file, int line, int level, const char *fmt,...)
        bvsnprintf(buf+len, sizeof(buf)-len, (char *)fmt, arg_ptr);
        va_end(arg_ptr);
 
-       /*
-        * Used the "trace on" command in the console to turn on
-        *  output to the trace file.  "trace off" will close the file.
-        */
-       if (trace) {
-          if (!trace_fd) {
-             char fn[200];
-             bsnprintf(fn, sizeof(fn), "%s/%s.trace", working_directory ? working_directory : "./", my_name);
-             trace_fd = fopen(fn, "a+b");
-          }
-          if (trace_fd) {
-             fputs(buf, trace_fd);
-             fflush(trace_fd);
-          } else {
-             /* Some problem, turn off tracing */
-             trace = false;
-          }
-       } else {   /* not tracing */
-          fputs(buf, stdout);
-          fflush(stdout);
-       }
+       pt_out(buf);
     }
 }
 
@@ -964,8 +1013,8 @@ p_msg(const char *file, int line, int level, const char *fmt,...)
     va_start(arg_ptr, fmt);
     bvsnprintf(buf+len, sizeof(buf)-len, (char *)fmt, arg_ptr);
     va_end(arg_ptr);
-    fputs(buf, stdout);
-    fflush(stdout);
+
+    pt_out(buf);     
 }
 
 
@@ -1324,8 +1373,6 @@ int Mmsg(POOL_MEM &pool_buf, const char *fmt, ...)
 }
 
 
-static pthread_mutex_t msg_queue_mutex = PTHREAD_MUTEX_INITIALIZER;
-
 /*
  * We queue messages rather than print them directly. This
  *  is generally used in low level routines (msg handler, bnet)
@@ -1360,16 +1407,15 @@ void Qmsg(JCR *jcr, int type, utime_t mtime, const char *fmt,...)
    if (!jcr) {
       jcr = get_jcr_from_tsd();
    }
-   /* If no jcr or dequeuing send to daemon to avoid recursion */
-   if ((jcr && !jcr->msg_queue) || !jcr || jcr->dequeuing) {
-      /* jcr==NULL => daemon message, safe to send now */
-      Jmsg(jcr, item->type, item->mtime, "%s", item->msg);
+   /* If no jcr or no queue or dequeuing send to syslog */
+   if (!jcr || !jcr->msg_queue || jcr->dequeuing_msgs) {
+      syslog(LOG_DAEMON|LOG_ERR, "%s", item->msg);
       free(item);
    } else {
       /* Queue message for later sending */
-      P(msg_queue_mutex);
+      P(jcr->msg_queue_mutex);
       jcr->msg_queue->append(item);
-      V(msg_queue_mutex);
+      V(jcr->msg_queue_mutex);
    }
    free_memory(pool_buf);
 }
@@ -1380,19 +1426,18 @@ void Qmsg(JCR *jcr, int type, utime_t mtime, const char *fmt,...)
 void dequeue_messages(JCR *jcr)
 {
    MQUEUE_ITEM *item;
-   P(msg_queue_mutex);
    if (!jcr->msg_queue) {
-      goto bail_out;
+      return;
    }
-   jcr->dequeuing = true;
+   P(jcr->msg_queue_mutex);
+   jcr->dequeuing_msgs = true;
    foreach_dlist(item, jcr->msg_queue) {
       Jmsg(jcr, item->type, item->mtime, "%s", item->msg);
    }
+   /* Remove messages just sent */
    jcr->msg_queue->destroy();
-   jcr->dequeuing = false;
-
-bail_out:
-   V(msg_queue_mutex);
+   jcr->dequeuing_msgs = false;
+   V(jcr->msg_queue_mutex);
 }