2007-10-02 Frank Ch. Eigler <fche@redhat.com>

	PR 5078
	* tapsets.cxx (be_derived_probe): Rework to add error probe support.
	Emit probe description array in C for traversal by generated code.
	* register_standard_tapsets: Add error probes.
	* stapprobes.5.in: Document.
	* translate.cxx (emit_module_init): Handle errors that may occur
	during begin probes.
	(emit_module_exit): Use schedule() rather than cpu_relax() during
	shutdown synchronization wait loop.
	* staptree.cxx (probe::printsig): Put multiple probe points on same
	line.

2007-10-02  Frank Ch. Eigler  <fche@redhat.com>

	PR 5078
	* semok/twentysix.stp, systemtap.base/beginenderror.*: New tests.

9 files changed, 172 insertions, 60 deletions

diff --git a/ChangeLog b/ChangeLog
index a855e857..8aa93076 100644
--- a/ChangeLog
+++ b/ChangeLog
@@ -1,3 +1,17 @@
+2007-10-02  Frank Ch. Eigler  <fche@redhat.com>
+
+	PR 5078
+	* tapsets.cxx (be_derived_probe): Rework to add error probe support.
+	Emit probe description array in C for traversal by generated code.
+	* register_standard_tapsets: Add error probes.
+	* stapprobes.5.in: Document.
+	* translate.cxx (emit_module_init): Handle errors that may occur
+	during begin probes.
+	(emit_module_exit): Use schedule() rather than cpu_relax() during
+	shutdown synchronization wait loop.
+	* staptree.cxx (probe::printsig): Put multiple probe points on same
+	line.
+
 2007-09-28  Frank Ch. Eigler  <fche@elastic.org>
 
 	* Makefile.in: Regenerated from Jim Keniston's uprobes Makefile.am
diff --git a/stapprobes.5.in b/stapprobes.5.in
index efefc1bf..536991ec 100644
--- a/stapprobes.5.in
+++ b/stapprobes.5.in
@@ -53,7 +53,7 @@ wildcards or aliases), and all them are then probed.  A probe
 declaration may also contain several comma-separated specifications,
 all of which are probed.
 
-.SS BEGIN/END
+.SS BEGIN/END/ERROR
 
 The probe points
 .IR begin " and " end
@@ -80,6 +80,16 @@ increasing order, and the order between handlers with the same sequence
 number is unspecified.  When "begin" or "end" are given without a
 sequence, they are effectively sequence zero.
 
+The
+.IR error
+probe point is similar to the 
+.IR end
+probe, except that each such probe handler run when the session and
+after errors having occurred.  In such cases, "end" probes are
+skipped, but each "error" prober is still attempted.  This kind of
+probe can be used to clean up or emit a final gasp message.  It may
+also be numerically parametrized to set a sequence.
+
 .SS NEVER
 The probe point
 .IR never
diff --git a/staptree.cxx b/staptree.cxx
index a1a8211b..ccfc8a8d 100644
--- a/staptree.cxx
+++ b/staptree.cxx
@@ -879,7 +879,7 @@ void probe::printsig (ostream& o) const
 {
   for (unsigned i=0; i<locations.size(); i++)
     {
-      if (i > 0) o << "," << endl;
+      if (i > 0) o << ",";
       locations[i]->print (o);
     }
 }
diff --git a/tapsets.cxx b/tapsets.cxx
index 1943b8af..c4d6962e 100644
--- a/tapsets.cxx
+++ b/tapsets.cxx
@@ -69,24 +69,31 @@ public:
 
 
 // ------------------------------------------------------------------------
-// begin/end probes are run right during registration / deregistration
+// begin/end/error probes are run right during registration / deregistration
 // ------------------------------------------------------------------------
 
+enum be_t { BEGIN, END, ERROR };
+
 struct be_derived_probe: public derived_probe
 {
-  bool begin;
+  be_t type;
   int64_t priority;
 
-  be_derived_probe (probe* p, bool b, int64_t pr):
-    derived_probe (p), begin (b), priority (pr) {}
-  be_derived_probe (probe* p, probe_point* l, bool b, int64_t pr):
-    derived_probe (p, l), begin (b), priority (pr) {}
+  be_derived_probe (probe* p, be_t t, int64_t pr):
+    derived_probe (p), type (t), priority (pr) {}
+  be_derived_probe (probe* p, probe_point* l, be_t t, int64_t pr):
+    derived_probe (p, l), type (t), priority (pr) {}
 
   void join_group (systemtap_session& s);
 
   static inline bool comp(be_derived_probe const *a,
 			  be_derived_probe const *b)
-  { return a->priority < b->priority; }
+  {
+    // This allows the BEGIN/END/ERROR probes to intermingle.
+    // But that's OK - they're always treversed with a nested
+    // "if (type==FOO)" conditional.
+    return a->priority < b->priority; 
+  }
 
   bool needs_global_locks () { return false; }
   // begin/end probes don't need locks around global variables, since
@@ -105,8 +112,10 @@ public:
 
 struct be_builder: public derived_probe_builder
 {
-  bool begin;
-  be_builder(bool b) : begin(b) {}
+  be_t type;
+
+  be_builder(be_t t) : type(t) {}
+
   virtual void build(systemtap_session &,
 		     probe * base,
 		     probe_point * location,
@@ -114,11 +123,12 @@ struct be_builder: public derived_probe_builder
 		     vector<derived_probe *> & finished_results)
   {
     int64_t priority;
-    if ((begin && !get_param(parameters, "begin", priority))
-	|| (!begin && !get_param(parameters, "end", priority)))
+    if ((type == BEGIN && !get_param(parameters, "begin", priority)) ||
+        (type == END && !get_param(parameters, "end", priority)) ||
+        (type == ERROR && !get_param(parameters, "error", priority)))
       priority = 0;
     finished_results.push_back(
-	new be_derived_probe(base, location, begin, priority));
+	new be_derived_probe(base, location, type, priority));
   }
 };
 
@@ -293,20 +303,52 @@ be_derived_probe_group::emit_module_decls (systemtap_session& s)
   if (probes.empty()) return;
 
   s.op->newline() << "/* ---- begin/end probes ---- */";
-  s.op->newline() << "void enter_begin_probe (void (*fn)(struct context*)) {";
+  s.op->newline() << "void enter_begin_probe (void (*fn)(struct context*), const char* pp) {";
   s.op->indent(1);
   common_probe_entryfn_prologue (s.op, "STAP_SESSION_STARTING", false, true);
-  s.op->newline() << "c->probe_point = \"begin\";";
+  s.op->newline() << "c->probe_point = pp;";
   s.op->newline() << "(*fn) (c);";
   common_probe_entryfn_epilogue (s.op, false, true);
   s.op->newline(-1) << "}";
-  s.op->newline() << "void enter_end_probe (void (*fn)(struct context*)) {";
+
+  s.op->newline() << "void enter_end_probe (void (*fn)(struct context*), const char* pp) {";
   s.op->indent(1);
   common_probe_entryfn_prologue (s.op, "STAP_SESSION_STOPPING", false, true);
-  s.op->newline() << "c->probe_point = \"end\";";
+  s.op->newline() << "c->probe_point = pp;";
+  s.op->newline() << "(*fn) (c);";
+  common_probe_entryfn_epilogue (s.op, false, true);
+  s.op->newline(-1) << "}";
+
+  s.op->newline() << "void enter_error_probe (void (*fn)(struct context*), const char* pp) {";
+  s.op->indent(1);
+  common_probe_entryfn_prologue (s.op, "STAP_SESSION_ERROR", false, true);
+  s.op->newline() << "c->probe_point = pp;";
   s.op->newline() << "(*fn) (c);";
   common_probe_entryfn_epilogue (s.op, false, true);
   s.op->newline(-1) << "}";
+
+  s.op->newline() << "struct stap_be_probe {";
+  s.op->newline(1) << "void (*ph)(struct context*);";
+  s.op->newline() << "const char* pp;";
+  s.op->newline() << "int type;";
+  s.op->newline(-1) << "} stap_be_probes[] = {";
+  s.op->indent(1);
+
+  // NB: We emit the table in sorted order here, so we don't have to
+  // store the priority numbers as integers and sort at run time.
+
+  sort(probes.begin(), probes.end(), be_derived_probe::comp);
+
+  for (unsigned i=0; i < probes.size(); i++)
+    {
+      s.op->newline () << "{"; 
+      s.op->line() << " .pp=" 
+                   << lex_cast_qstring (*probes[i]->sole_location()) << ",";
+      s.op->line() << " .ph=&" << probes[i]->name << ",";
+      s.op->line() << " .type=" << probes[i]->type;
+      s.op->line() << " },";
+    }
+  s.op->newline(-1) << "};";
 }
 
 void
@@ -314,19 +356,17 @@ be_derived_probe_group::emit_module_init (systemtap_session& s)
 {
   if (probes.empty()) return;
 
-  bool have_begin_probes = false;
-  sort(probes.begin(), probes.end(), be_derived_probe::comp);
-  for (unsigned i=0; i < probes.size (); i++)
-    if (probes[i]->begin)
-      {
-	have_begin_probes = true;
-	s.op->newline() << "enter_begin_probe (& " << probes[i]->name << ");";
-      }
-
-  // If any of the begin probes signaled an error, indicate
-  // failure to the rest of systemtap_module_init.
-  if (have_begin_probes)
-    s.op->newline() << "rc = (atomic_read (&session_state) == STAP_SESSION_ERROR);";
+  s.op->newline() << "for (i=0; i<" << probes.size() << "; i++) {";
+  s.op->newline(1) << "struct stap_be_probe* stp = & stap_be_probes [i];";
+  s.op->newline() << "if (stp->type != " << BEGIN << ") continue;";
+  s.op->newline() << "enter_begin_probe (stp->ph, stp->pp);";
+  s.op->newline() << "/* rc = 0; */";
+  // NB: begin probes that cause errors do not constitute registration
+  // failures.  An error message will probably get printed and if
+  // MAXERRORS was left at 1, we'll get an stp_exit.  The
+  // error-handling probes will be run during the ordinary
+  // unregistration phase.
+  s.op->newline(-1) << "}";
 }
 
 void
@@ -334,10 +374,17 @@ be_derived_probe_group::emit_module_exit (systemtap_session& s)
 {
   if (probes.empty()) return;
 
-  sort(probes.begin(), probes.end(), be_derived_probe::comp);
-  for (unsigned i=0; i < probes.size (); i++)
-    if (! probes[i]->begin) // note polarity
-      s.op->newline() << "enter_end_probe (& " << probes[i]->name << ");";
+  s.op->newline() << "for (i=0; i<" << probes.size() << "; i++) {";
+  s.op->newline(1) << "struct stap_be_probe* stp = & stap_be_probes [i];";
+  s.op->newline() << "if (stp->type != " << END << ") continue;";
+  s.op->newline() << "enter_end_probe (stp->ph, stp->pp);";
+  s.op->newline(-1) << "}";
+
+  s.op->newline() << "for (i=0; i<" << probes.size() << "; i++) {";
+  s.op->newline(1) << "struct stap_be_probe* stp = & stap_be_probes [i];";
+  s.op->newline() << "if (stp->type != " << ERROR << ") continue;";
+  s.op->newline() << "enter_error_probe (stp->ph, stp->pp);";
+  s.op->newline(-1) << "}";
 }
 
 
@@ -6246,10 +6293,12 @@ perfmon_derived_probe_group::emit_module_init (translator_output* o)
 void
 register_standard_tapsets(systemtap_session & s)
 {
-  s.pattern_root->bind("begin")->bind(new be_builder(true));
-  s.pattern_root->bind_num("begin")->bind(new be_builder(true));
-  s.pattern_root->bind("end")->bind(new be_builder(false));
-  s.pattern_root->bind_num("end")->bind(new be_builder(false));
+  s.pattern_root->bind("begin")->bind(new be_builder(BEGIN));
+  s.pattern_root->bind_num("begin")->bind(new be_builder(BEGIN));
+  s.pattern_root->bind("end")->bind(new be_builder(END));
+  s.pattern_root->bind_num("end")->bind(new be_builder(END));
+  s.pattern_root->bind("error")->bind(new be_builder(ERROR));
+  s.pattern_root->bind_num("error")->bind(new be_builder(ERROR));
 
   s.pattern_root->bind("never")->bind(new never_builder());
 
diff --git a/testsuite/ChangeLog b/testsuite/ChangeLog
index a9f4488f..270b2caf 100644
--- a/testsuite/ChangeLog
+++ b/testsuite/ChangeLog
@@ -1,3 +1,8 @@
+2007-10-02  Frank Ch. Eigler  <fche@redhat.com>
+
+	PR 5078
+	* semok/twentysix.stp, systemtap.base/beginenderror.*: New tests.
+
 2007-10-02  William Cohen  <wcohen@redhat.com>
 
 	* systemtap.syscall/signal.c:
diff --git a/testsuite/semok/twentysix.stp b/testsuite/semok/twentysix.stp
new file mode 100644
index 00000000..b18a490d
--- /dev/null
+++ b/testsuite/semok/twentysix.stp
@@ -0,0 +1,4 @@
+#! stap -p2
+
+probe begin,begin(10),end,end(10),error,error(10) 
+{ println (pp()) }
diff --git a/testsuite/systemtap.base/beginenderror.exp b/testsuite/systemtap.base/beginenderror.exp
new file mode 100644
index 00000000..9ad596cb
--- /dev/null
+++ b/testsuite/systemtap.base/beginenderror.exp
@@ -0,0 +1,22 @@
+set test "beginenderror"
+
+if {![installtest_p]} {untested $test; return}
+
+spawn stap $srcdir/$subdir/beginenderror.stp
+set ok1 0
+set ok2 0
+set ko 0
+expect {
+    -re {^ERROR:[^\r\n]*\r\n} { incr ok1; exp_continue }
+    -re {^WARNING:[^\r\n]*\r\n} { incr ok1; exp_continue }
+    -re {^ok[^\r\n]*\r\n} { incr ok2; exp_continue }
+    -re {^ko[^\r\n]*\r\n} { incr ko; exp_continue }
+    timeout { fail "$test (timeout)" }
+    eof { }
+}
+wait
+if {$ok1 == 4 && $ok2 == 3 && $ko == 0} then {
+    pass "$test ($ok1 $ok2 $ko)"
+} else {
+    fail "$test ($ok1 $ok2 $ko)" 
+}
diff --git a/testsuite/systemtap.base/beginenderror.stp b/testsuite/systemtap.base/beginenderror.stp
new file mode 100644
index 00000000..8d0772cd
--- /dev/null
+++ b/testsuite/systemtap.base/beginenderror.stp
@@ -0,0 +1,5 @@
+probe begin { println ("ok ".pp()) println (1/0) println ("ko ".pp()) } # handler should unwind before seemenot message
+probe begin(10) { println ("ko ".pp()) } # this late begin probe shouldn't run, ERROR
+probe end, end(10) { println ("ko ".pp()) } # end probes won't be run due to ERROR state
+probe error, error(10) { println ("ok ".pp()) } # these handlers should be run
+probe error(20),error(30) { println (1/0) println ("ko ".pp()) } # likewise
diff --git a/translate.cxx b/translate.cxx
index 7ff2bae6..f2351b9f 100644
--- a/translate.cxx
+++ b/translate.cxx
@@ -1144,23 +1144,6 @@ c_unparser::emit_module_init ()
     }
   o->newline() << "#endif";
 
-  for (unsigned i=0; i<g.size(); i++)
-    {
-      g[i]->emit_module_init (*session);
-      // NB: this gives O(N**2) amount of code, but luckily there
-      // are only seven or eight derived_probe_groups, so it's ok.
-      o->newline() << "if (rc) {";
-      o->newline(1) << "_stp_error (\"probe %s registration error (rc %d)\", probe_point, rc);";
-      // NB: we need to be in the error state so timers can shutdown cleanly,
-      // and so end probes don't run.
-      o->newline() << "atomic_set (&session_state, STAP_SESSION_ERROR);";
-      if (i>0)
-        for (int j=i-1; j>=0; j--)
-          g[j]->emit_module_exit (*session);
-      o->newline() << "goto out;";
-      o->newline(-1) << "}";
-    }
-
   // Print a message to the kernel log about this module.  This is
   // intended to help debug problems with systemtap modules.
   o->newline() << "printk (KERN_DEBUG \"%s: "
@@ -1179,9 +1162,30 @@ c_unparser::emit_module_init ()
 	       << ", (unsigned long) _stp_allocated_memory"
                << ");";
 
+  // Run all probe registrations.  This actually runs begin probes.
+
+  for (unsigned i=0; i<g.size(); i++)
+    {
+      g[i]->emit_module_init (*session);
+      // NB: this gives O(N**2) amount of code, but luckily there
+      // are only seven or eight derived_probe_groups, so it's ok.
+      o->newline() << "if (rc) {";
+      o->newline(1) << "_stp_error (\"probe %s registration error (rc %d)\", probe_point, rc);";
+      // NB: we need to be in the error state so timers can shutdown cleanly,
+      // and so end probes don't run.  OTOH, error probes can run.
+      o->newline() << "atomic_set (&session_state, STAP_SESSION_ERROR);";
+      if (i>0)
+        for (int j=i-1; j>=0; j--)
+          g[j]->emit_module_exit (*session);
+      o->newline() << "goto out;";
+      o->newline(-1) << "}";
+    }
+
   // All registrations were successful.  Consider the system started.
-  o->newline() << "atomic_set (&session_state, STAP_SESSION_RUNNING);";
-  o->newline() << "return 0;";
+  o->newline() << "if (atomic_read (&session_state) == STAP_SESSION_STARTING)";
+  // NB: only other valid state value is ERROR, in which case we don't 
+  o->newline(1) << "atomic_set (&session_state, STAP_SESSION_RUNNING);";
+  o->newline(-1) << "return 0;";
 
   // Error handling path; by now all partially registered probe groups
   // have been unregistered.
@@ -1237,8 +1241,7 @@ c_unparser::emit_module_exit ()
   o->newline(1) << "if (cpu_possible (i) && " 
                 << "atomic_read (& ((struct context *)per_cpu_ptr(contexts, i))->busy)) "
                 << "holdon = 1;";
-  o->newline () << "cpu_relax ();";
-  // o->newline(-1) << "if (holdon) msleep (5);";
+  o->newline () << "schedule ();";
   o->newline(-1) << "} while (holdon);";
   o->newline(-1);
   // XXX: might like to have an escape hatch, in case some probe is