Files pan2.svn-orig/core and pan2.svn-decoder-threads-works!/core differ
diff --exclude-from=diff-excludes -urN pan2.svn-orig/pan/data-impl/add-server.cc pan2.svn-decoder-threads-works!/pan/data-impl/add-server.cc
--- pan2.svn-orig/pan/data-impl/add-server.cc	2007-03-20 06:17:58.000000000 -0400
+++ pan2.svn-decoder-threads-works!/pan/data-impl/add-server.cc	2007-03-19 02:40:39.000000000 -0400
@@ -56,8 +56,7 @@
 
   // initialize the queue
   TaskArchive null_task_archive;
-  GIOChannelSocket::Creator _socket_creator;
-  Queue queue (data, null_task_archive, &_socket_creator, true);
+  Queue queue (data, null_task_archive, true);
   queue.add_task (new TaskGroups (data, servername));
 
   // start the event loop...
diff --exclude-from=diff-excludes -urN pan2.svn-orig/pan/general/Makefile.am pan2.svn-decoder-threads-works!/pan/general/Makefile.am
--- pan2.svn-orig/pan/general/Makefile.am	2007-03-20 06:17:59.000000000 -0400
+++ pan2.svn-decoder-threads-works!/pan/general/Makefile.am	2007-03-18 17:52:50.000000000 -0400
@@ -8,6 +8,7 @@
  file-util.cc \
  log.cc \
  line-reader.cc \
+ locking.cc \
  progress.cc \
  quark.cc \
  string-view.cc \
@@ -20,6 +21,7 @@
  file-util.h \
  foreach.h \
  line-reader.h \
+ locking.h \
  log.h \
  map-vector.h \
  messages.h \
diff --exclude-from=diff-excludes -urN pan2.svn-orig/pan/general/debug.cc pan2.svn-decoder-threads-works!/pan/general/debug.cc
--- pan2.svn-orig/pan/general/debug.cc	2007-03-20 06:17:59.000000000 -0400
+++ pan2.svn-decoder-threads-works!/pan/general/debug.cc	2007-03-20 02:27:11.000000000 -0400
@@ -4,4 +4,5 @@
 namespace pan
 {
   bool _debug_flag = false;
+  bool _debug_verbose_flag = false;
 }
diff --exclude-from=diff-excludes -urN pan2.svn-orig/pan/general/debug.h pan2.svn-decoder-threads-works!/pan/general/debug.h
--- pan2.svn-orig/pan/general/debug.h	2007-03-20 06:17:59.000000000 -0400
+++ pan2.svn-decoder-threads-works!/pan/general/debug.h	2007-03-20 02:31:05.000000000 -0400
@@ -25,6 +25,7 @@
 namespace pan
 {
   extern bool _debug_flag;
+  extern bool _debug_verbose_flag;
 }
 
 #define LINE_ID '(' << __FILE__ << ':' << __LINE__ << ':' << __func__ << ')'
@@ -35,4 +36,10 @@
       std::cerr << LINE_ID << ' ' << A << '\n'; \
   } while (0)
 
+#define debug_v(A) \
+  do { \
+    if (_debug_verbose_flag) \
+      std::cerr << LINE_ID << ' ' << A << '\n'; \
+  } while (0)
+
 #endif
diff --exclude-from=diff-excludes -urN pan2.svn-orig/pan/general/locking.cc pan2.svn-decoder-threads-works!/pan/general/locking.cc
--- pan2.svn-orig/pan/general/locking.cc	1969-12-31 19:00:00.000000000 -0500
+++ pan2.svn-decoder-threads-works!/pan/general/locking.cc	2007-03-18 19:48:21.000000000 -0400
@@ -0,0 +1,128 @@
+/* -*- Mode: C++; tab-width: 2; indent-tabs-mode: nil; c-basic-offset: 2 -*- */
+/*
+ * Pan - A Newsreader for Gtk+
+ * Copyright (C) 2007  Calin Culianu <calin@ajvar.org>
+ *
+ * This program is free software; you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation; version 2 of the License.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program; if not, write to the Free Software
+ * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
+ */
+#include "locking.h"
+#include <glib.h>
+#include <cassert>
+
+namespace pan 
+{
+  struct Mutex::Impl {     GMutex *mut;       };
+  
+  Lockable::Lockable() { if (!g_thread_supported ()) g_thread_init (NULL); }
+
+  Mutex::Mutex() { p = new Impl; p->mut = g_mutex_new(); islocked = false;}
+  Mutex::~Mutex() { g_mutex_free(p->mut); delete p;  p = 0; }
+  
+  void Mutex::lock() { g_mutex_lock(p->mut); islocked = true; }
+  void Mutex::unlock() { islocked = false; g_mutex_unlock(p->mut); }
+  bool Mutex::tryLock() { 
+    bool ret = g_mutex_trylock(p->mut); 
+    if (ret) islocked = true; 
+    return ret; 
+  }
+
+  struct RWLock::Impl {     GStaticRWLock rw;  };
+  
+  RWLock::RWLock() 
+    : p(new Impl), isreadlocked(false), iswritelocked(false)
+  { 
+    g_static_rw_lock_init(&p->rw); 
+  }
+  RWLock::~RWLock() { g_static_rw_lock_free(&p->rw); delete p;  p = 0; }
+  
+  void RWLock::lock() 
+  { 
+    g_static_rw_lock_reader_lock(&p->rw); 
+    assert(!iswritelocked);    
+    isreadlocked = true; 
+  }
+
+  void RWLock::unlock() 
+  { 
+    assert(isreadlocked);
+    assert(!iswritelocked);
+    isreadlocked = false; 
+    g_static_rw_lock_reader_unlock(&p->rw); 
+  }
+
+  bool RWLock::tryLock() 
+  { 
+    bool ret = g_static_rw_lock_reader_trylock(&p->rw); 
+    if (ret) { isreadlocked = true;  assert(!iswritelocked); }
+    return ret; 
+  }
+
+  void RWLock::writeLock() 
+  { 
+    g_static_rw_lock_writer_lock(&p->rw); 
+    iswritelocked = true; 
+    assert(!isreadlocked);
+  }
+
+  void RWLock::writeUnlock() 
+  { 
+    assert(iswritelocked);
+    assert(!isreadlocked);
+    iswritelocked = false; 
+    g_static_rw_lock_reader_unlock(&p->rw); 
+  }
+
+  bool RWLock::tryWriteLock() 
+  { 
+    bool ret = g_static_rw_lock_writer_trylock(&p->rw); 
+    if (ret) { iswritelocked = true; assert(!isreadlocked); }
+    return ret; 
+  }
+  
+  struct Cond::Impl { GCond *c; };
+
+  Cond :: Cond() { p = new Impl;  p->c = g_cond_new(); }
+  Cond ::  ~Cond() { g_cond_free(p->c); delete p; p = 0; }
+  
+    // signal sleepers
+
+  void 
+  Cond :: signal() { g_cond_signal(p->c); }
+  void
+  Cond :: broadcast() { g_cond_broadcast(p->c); }
+  
+    // be a sleeper 
+
+  void Cond :: wait(Mutex &mut) 
+  { 
+    assert(mut.isLocked());
+    g_cond_wait(p->c, mut.p->mut); 
+  }
+  
+  /** Like wait() but optionally can time out and return false 
+        if the condition was not signalled.  Returns true otherwise.  In 
+        either case the mutex should have been acquired before entering this
+        function and will be locked upon return. */
+  bool Cond :: timedWait(Mutex &mut, unsigned time_msecs) 
+  {
+    assert(mut.isLocked());
+    GTimeVal tv;
+    g_get_current_time(&tv);
+    g_time_val_add(&tv, ((glong)time_msecs)*1000L);
+    return g_cond_timed_wait(p->c, mut.p->mut, &tv); 
+  }
+  
+
+}
+
diff --exclude-from=diff-excludes -urN pan2.svn-orig/pan/general/locking.h pan2.svn-decoder-threads-works!/pan/general/locking.h
--- pan2.svn-orig/pan/general/locking.h	1969-12-31 19:00:00.000000000 -0500
+++ pan2.svn-decoder-threads-works!/pan/general/locking.h	2007-03-18 19:49:17.000000000 -0400
@@ -0,0 +1,136 @@
+/* -*- Mode: C++; tab-width: 2; indent-tabs-mode: nil; c-basic-offset: 2 -*- */
+/*
+ * Pan - A Newsreader for Gtk+
+ * Copyright (C) 2007  Calin Culianu <calin@ajvar.org>
+ *
+ * This program is free software; you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation; version 2 of the License.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program; if not, write to the Free Software
+ * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
+ */
+#ifndef pan_colon_colon_locking_H
+#define pan_colon_colon_locking_H
+
+namespace pan
+{
+
+  ///< pure virtual lockable interface for some generic algorithms...?
+  class Lockable
+  {
+  public:
+    Lockable(); ///< c'tor here needed because it may do some initialization application-wide
+    virtual ~Lockable() {}
+    virtual void lock() = 0;
+    virtual void unlock() = 0;
+    virtual bool isLocked() const = 0;
+    virtual bool tryLock() = 0;
+  };
+
+  ///< A wrapper around GMutex that is more C++-ey
+  class Mutex : public Lockable
+  {
+  public:
+    Mutex();
+    ~Mutex();
+    void lock(); ///< lock the mutex -- blocks until it is acquired
+    void unlock(); ///< unlock the mutex -- may wake 1 other thread waiting on it
+    bool isLocked() const { return islocked; }
+    bool tryLock(); ///< may lock the mutex -- does not block and returns false if it could not lock the mutex, or true if it could
+  private:
+    struct Impl; ///< needed this type so we don't put GMutex here and thus have to pull in glib.h, etc
+    Impl *p; 
+    volatile bool islocked;
+    friend class Cond;
+  };
+  
+  /** A wrapper around GStaticRWLock.  Note that lock() acquires the read lock
+   *  and writeLock() acquires the write lock.
+   */
+  class RWLock : public Lockable
+  {
+    RWLock();
+    ~RWLock();
+    
+    // lockable interface methods ---
+
+    void lock(); ///< acquire the readlock -- blocks until it is acquired
+    void unlock(); ///< releasthe the readlock, waking any potential sleepers
+    bool isLocked() const { return isreadlocked; } ///< true iff readlock is acquired
+    bool tryLock(); ///< try to acquire read lock -- does not block and returns false if it could not acquire, or true if it could
+
+    // methods specific to RWLock
+
+    void writeLock(); ///< acquire the write lock -- may block
+    void writeUnlock(); ///< release the write lock -- may wake sleepers
+    bool isWriteLocked() const { return iswritelocked; } ///< true iff write lock is acquired
+    bool tryWriteLock(); ///< true iff writelock was acquired, false otherwise -- does not block
+
+  private:
+    struct Impl; ///< needed this type so we don't put GStaicRWLock here and thus have to pull in glib.h, etc
+    Impl *p; 
+    volatile bool isreadlocked, iswritelocked;    
+  };
+
+  /** Convenience class that automatically locks and unlocks a lockable.
+   *  Locking is done in the class constructor, and unlocking is done in the
+   *  destructor.  This is a convenience for automatic locking and unlocking
+   *  within a C++ language scope. 
+   */
+  class AutoLocker
+  {
+  public:
+    /// construct from a reference to a Mutex, locks mutex
+    AutoLocker(Lockable & l) : lock(l) { lock.lock(); }
+    /// construct from a pointer to Mutex, locks mutex
+    AutoLocker(Lockable *l) : lock(*l) { lock.lock(); }
+    /// unlocks mutex
+    ~AutoLocker() { lock.unlock(); }
+  private:
+    Lockable & lock; ///< our mutex or whatever that we auto-lock/unlock
+  };  
+
+  typedef class AutoLocker MutexLocker; ///< sometimes it's easier for readability to call it a MutexLocker
+
+  
+  /// wraps GCond, your basic condition functionality like in glib, pthreads, et al
+  class Cond
+  {
+  public:
+    Cond();
+    ~Cond();
+    
+    // signal sleepers
+
+    void signal(); ///< signal condition, waking at most 1 sleeper
+    void broadcast(); ///< wake any and all sleepers
+
+    // be a sleeper 
+
+    /** wait on a condition --
+        mutex is atomically released before sleeping then
+        acquired again when this returns */
+    void wait(Mutex &mut); 
+    void wait(Mutex *m) { wait(*m); }
+    /** Like wait() but optionally can time out and return false 
+        if the condition was not signalled.  Returns true otherwise.  In 
+        either case the mutex should have been acquired before entering this
+        function and will be locked upon return. */
+    bool timedWait(Mutex &mut, unsigned time_msecs); 
+    /// like above function but takes a pointer arg
+    bool timedWait(Mutex *m, unsigned t) { return timedWait(*m,t); }
+
+  private:
+    struct Impl;
+    Impl *p;
+  };
+}
+
+#endif
diff --exclude-from=diff-excludes -urN pan2.svn-orig/pan/general/progress.h pan2.svn-decoder-threads-works!/pan/general/progress.h
--- pan2.svn-orig/pan/general/progress.h	2007-03-20 06:17:59.000000000 -0400
+++ pan2.svn-decoder-threads-works!/pan/general/progress.h	2007-03-19 02:14:41.000000000 -0400
@@ -56,10 +56,6 @@
 
     private:
 
-      typedef std::set<Listener*> listeners_t;
-      typedef Progress::listeners_t::const_iterator listeners_cit;
-      listeners_t _listeners;
-
       void fire_pulse ();
       void fire_percentage (int p);
       void fire_status (const StringView& msg);
@@ -68,6 +64,10 @@
 
     protected:
 
+      typedef std::set<Listener*> listeners_t;
+      typedef Progress::listeners_t::const_iterator listeners_cit;
+      listeners_t _listeners;
+
       std::string _description; // used for default describe()
       std::string _status_text; // the last status text emitted
       std::vector<std::string> _errors; // the emitted error strings
diff --exclude-from=diff-excludes -urN pan2.svn-orig/pan/gui/gui.cc pan2.svn-decoder-threads-works!/pan/gui/gui.cc
--- pan2.svn-orig/pan/gui/gui.cc	2007-03-20 06:17:57.000000000 -0400
+++ pan2.svn-decoder-threads-works!/pan/gui/gui.cc	2007-03-20 05:48:01.000000000 -0400
@@ -33,6 +33,7 @@
 #include <pan/tasks/task-groups.h>
 #include <pan/tasks/task-xover.h>
 #include <pan/tasks/nzb.h>
+#include <pan/tasks/worker-pool.h>
 #include <pan/icons/pan-pixbufs.h>
 #include "actions.h"
 #include "body-pane.h"
@@ -317,6 +318,17 @@
   _prefs.add_listener (this);
 
   gtk_accel_map_load (get_accel_filename().c_str());
+
+  { // make sure taskbar views have the right tasks in them -- this is because when Pan first starts the active tasks are already running
+    Queue::task_states_t task_states;
+    queue.get_all_task_states(task_states);    
+    foreach(Queue::tasks_t, task_states.tasks, it) {
+      Queue::TaskState s = task_states.get_state(*it);
+      if (s == Queue::RUNNING || s == Queue::DECODING)
+        on_queue_task_active_changed (queue, *(*it), true);
+    }
+  }
+
 }
 
 namespace
@@ -1142,13 +1154,13 @@
 void GUI :: do_about_pan ()
 {
 #if GTK_CHECK_VERSION(2,6,0)
-  const gchar * authors [] = { "Charles Kerr", 0 };
+  const gchar * authors [] = { "Charles Kerr", "Calin Culianu <calin@ajvar.org>", 0 };
   GdkPixbuf * logo = gdk_pixbuf_new_from_inline(-1, icon_pan_about_logo, 0, 0);
   GtkAboutDialog * w (GTK_ABOUT_DIALOG (gtk_about_dialog_new ()));
   gtk_about_dialog_set_name (w, _("Pan"));
   gtk_about_dialog_set_version (w, PACKAGE_VERSION);
   gtk_about_dialog_set_comments (w, VERSION_TITLE);
-  gtk_about_dialog_set_copyright (w, _("Copyright © 2002-2006 Charles Kerr"));
+  gtk_about_dialog_set_copyright (w, _("Copyright © 2002-2006 Charles Kerr\nCopyright © 2007 Calin Culianu"));
   gtk_about_dialog_set_website (w, "http://pan.rebelbase.com/");
   gtk_about_dialog_set_logo (w, logo);
   gtk_about_dialog_set_license (w, LICENSE);
@@ -1168,7 +1180,7 @@
   gtk_box_pack_start (box, gtk_image_new_from_pixbuf (logo), false, false, PAD);
   gtk_box_pack_start (box, gtk_label_new("Pan " PACKAGE_VERSION), false, false, PAD);
   gtk_box_pack_start (box, gtk_label_new(VERSION_TITLE), false, false, 0);
-  gtk_box_pack_start (box, gtk_label_new(_("Copyright © 2002-2006 Charles Kerr")), false, false, 0);
+  gtk_box_pack_start (box, gtk_label_new(_("Copyright © 2002-2006 Charles Kerr\nCopyright © 2007 Calin Culianu")), false, false, 0);
   gtk_box_pack_start (box, gtk_label_new("http://pan.rebelbase.com/"), false, false, PAD);
   gtk_widget_show_all (dialog);
   g_signal_connect_swapped (dialog, "response", G_CALLBACK (gtk_widget_destroy), dialog);
@@ -1415,6 +1427,7 @@
 
 void GUI :: do_quit ()
 {
+  WorkerPool::cancelAllWorkers();
   gtk_main_quit ();
 }
 void GUI :: do_read_selected_group ()
diff --exclude-from=diff-excludes -urN pan2.svn-orig/pan/gui/pan.cc pan2.svn-decoder-threads-works!/pan/gui/pan.cc
--- pan2.svn-orig/pan/gui/pan.cc	2007-03-20 06:17:57.000000000 -0400
+++ pan2.svn-decoder-threads-works!/pan/gui/pan.cc	2007-03-20 02:28:25.000000000 -0400
@@ -60,6 +60,7 @@
 
   void mainloop_quit ()
   {
+    WorkerPool::cancelAllWorkers(); // tell any potentially running worker pool workers to stop asap
     if (nongui_gmainloop)
       g_main_loop_quit (nongui_gmainloop);
     else
@@ -92,6 +93,7 @@
 
   void destroy_cb (GtkWidget*w, gpointer user_data)
   {
+    WorkerPool::cancelAllWorkers(); // tell any potentially running worker pool workers to stop asap
     gtk_main_quit ();
   }
 
@@ -183,7 +185,6 @@
   bind_textdomain_codeset (GETTEXT_PACKAGE, "UTF-8");
   textdomain (GETTEXT_PACKAGE);
 
-  g_thread_init (0);
   gtk_init (&argc, &argv);
   g_mime_init (GMIME_INIT_FLAG_UTF8);
 
@@ -200,9 +201,10 @@
       url = tok;
     else if (!strcmp(tok,"--no-gui") || !strcmp(tok,"--nogui"))
       gui = false;
-    else if (!strcmp (tok, "--debug"))
-      _debug_flag = true;
-    else if (!strcmp (tok, "--nzb"))
+    else if (!strcmp (tok, "--debug")) { // do --debug --debug for verbose debug
+      if (_debug_flag) _debug_verbose_flag = true;
+      else _debug_flag = true;
+    } else if (!strcmp (tok, "--nzb"))
       nzb = true;
     else if (!strcmp (tok, "--version"))
       { std::cerr << "Pan " << VERSION << '\n'; return 0; }
@@ -244,8 +246,7 @@
     }
 
     // instantiate the queue...
-    GIOChannelSocket::Creator socket_creator;
-    Queue queue (data, data, &socket_creator, prefs.get_flag("work-online",true));
+    Queue queue (data, data, prefs.get_flag("work-online",true));
     g_timeout_add (5000, queue_upkeep_timer_cb, &queue);
 
     if (nzb)
diff --exclude-from=diff-excludes -urN pan2.svn-orig/pan/gui/task-pane.cc pan2.svn-decoder-threads-works!/pan/gui/task-pane.cc
--- pan2.svn-orig/pan/gui/task-pane.cc	2007-03-20 06:17:57.000000000 -0400
+++ pan2.svn-decoder-threads-works!/pan/gui/task-pane.cc	2007-03-20 05:07:12.000000000 -0400
@@ -210,11 +210,11 @@
   {
     Task * task (*it);
     const Queue::TaskState state (tasks.get_state (task));
-    if (state == Queue::RUNNING)
+    if (state == Queue::RUNNING || state == Queue::DECODING)
       ++running_count;
     else if (state == Queue::STOPPED)
       ++stopped_count;
-    else if (state == Queue::QUEUED)
+    else if (state == Queue::QUEUED || state == Queue::QUEUED_FOR_DECODE)
       ++queued_count;
 
     if (state==Queue::RUNNING || state==Queue::QUEUED)
@@ -315,6 +315,8 @@
     const char * state_str (0);
     switch (state) {
       case Queue::RUNNING:  state_str = _("Running"); break;
+      case Queue::DECODING: state_str = _("Decoding"); break;
+      case Queue::QUEUED_FOR_DECODE: state_str = _("Queued for Decode"); break;
       case Queue::QUEUED:   state_str = _("Queued"); break;
       case Queue::STOPPED:  state_str = _("Stopped"); break;
       case Queue::REMOVING: state_str = _("Removing"); break;
@@ -372,7 +374,7 @@
     }
 
     char * str (0);
-    if (state == Queue::RUNNING)
+    if (state == Queue::RUNNING || state == Queue::DECODING)
       str = g_markup_printf_escaped ("<b>%s</b>\n<small>%s</small>", description.c_str(), status.c_str());
     else
       str = g_markup_printf_escaped ("%s\n<small>%s</small>", description.c_str(), status.c_str());
diff --exclude-from=diff-excludes -urN pan2.svn-orig/pan/tasks/Makefile.am pan2.svn-decoder-threads-works!/pan/tasks/Makefile.am
--- pan2.svn-orig/pan/tasks/Makefile.am	2007-03-20 06:17:58.000000000 -0400
+++ pan2.svn-decoder-threads-works!/pan/tasks/Makefile.am	2007-03-19 15:17:14.000000000 -0400
@@ -3,8 +3,10 @@
 noinst_LIBRARIES = libtasks.a
 
 libtasks_a_SOURCES = \
+  globals.cc \
   task.cc \
   task-article.cc \
+  task-article-decoder.cc \
   task-groups.cc \
   task-post.cc \
   task-xover.cc \
@@ -14,15 +16,18 @@
   socket.cc \
   socket-impl-gio.cc \
   socket-impl-scripted.cc \
-  nntp-pool.cc
+  nntp-pool.cc \
+  worker-pool.cc
 
 noinst_HEADERS = \
   adaptable-set.cc \
   adaptable-set.h \
   defgroup.h \
+  globals.h \
   health.h \
   task.h \
   task-article.h \
+  task-article-decoder.h \
   task-groups.h \
   task-post.h \
   task-weak-ordering.h \
@@ -33,7 +38,8 @@
   socket.h \
   socket-impl-gio.h \
   socket-impl-scripted.h \
-  nntp-pool.h
+  nntp-pool.h \
+  worker-pool.h
 
 noinst_PROGRAMS = \
   adaptable-set-test \
diff --exclude-from=diff-excludes -urN pan2.svn-orig/pan/tasks/globals.cc pan2.svn-decoder-threads-works!/pan/tasks/globals.cc
--- pan2.svn-orig/pan/tasks/globals.cc	1969-12-31 19:00:00.000000000 -0500
+++ pan2.svn-decoder-threads-works!/pan/tasks/globals.cc	2007-03-21 19:18:05.000000000 -0400
@@ -0,0 +1,62 @@
+/* -*- Mode: C++; tab-width: 2; indent-tabs-mode: nil; c-basic-offset: 2 -*- */
+/*
+ * Pan - A Newsreader for Gtk+
+ * Copyright (C) 2007  Calin Culianu <calin@ajvar.org>
+ *
+ * This program is free software; you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation; version 2 of the License.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program; if not, write to the Free Software
+ * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
+ */
+#include "globals.h"
+
+namespace {
+
+  pan::WorkerPool *worker_pool = 0;
+  pan::GIOChannelSocket::Creator *socket_creator = 0;
+  pan::Mutex *mutex = 0;
+
+  struct Deleter { 
+    ~Deleter() {
+      // NB to noobies: it's ok to delete 0
+      delete worker_pool; worker_pool = 0;
+      delete socket_creator; socket_creator = 0;
+      delete mutex; mutex = 0;      
+    } 
+  } deleter; // d'tor called on app exit, which deletes heap-allocated objects
+
+}
+
+namespace pan
+{
+  namespace globals
+  {
+    WorkerPool & workerPool() 
+    { 
+      if (!worker_pool) 
+        // hard-coded parameters for the worker pool...
+        worker_pool = new WorkerPool(4, true); 
+      return *worker_pool;
+    }
+    
+    GIOChannelSocket::Creator & socketCreator() 
+    {
+      if (!socket_creator) socket_creator = new GIOChannelSocket::Creator;
+      return *socket_creator;
+    }
+
+    Mutex & mutex() 
+    {
+      if (!::mutex) ::mutex = new Mutex;
+      return *::mutex;
+    }
+  }
+}
diff --exclude-from=diff-excludes -urN pan2.svn-orig/pan/tasks/globals.h pan2.svn-decoder-threads-works!/pan/tasks/globals.h
--- pan2.svn-orig/pan/tasks/globals.h	1969-12-31 19:00:00.000000000 -0500
+++ pan2.svn-decoder-threads-works!/pan/tasks/globals.h	2007-03-21 19:18:05.000000000 -0400
@@ -0,0 +1,42 @@
+/* -*- Mode: C++; tab-width: 2; indent-tabs-mode: nil; c-basic-offset: 2 -*- */
+/*
+ * Pan - A Newsreader for Gtk+
+ * Copyright (C) 2007  Calin Culianu <calin@ajvar.org>
+ *
+ * This program is free software; you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation; version 2 of the License.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program; if not, write to the Free Software
+ * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
+ */
+#ifndef pan_colon_colon_globals_H
+#define pan_colon_colon_globals_H
+
+#include <pan/tasks/worker-pool.h>
+#include <pan/tasks/socket-impl-gio.h>
+#include <pan/general/locking.h>
+
+namespace pan
+{
+  /// some global instances that can be shared and used throughout the application
+  namespace globals
+  {
+    /// useful to dole out work to slave threads
+    extern WorkerPool & workerPool();
+
+    /// a pointer to this used to be passed around to objects -- better to just make it global
+    extern GIOChannelSocket::Creator & socketCreator();
+
+    /// mutex that can be used for very coarse-grained locking appliction-wide
+    extern Mutex & mutex();    
+  }
+}
+#endif
+
diff --exclude-from=diff-excludes -urN pan2.svn-orig/pan/tasks/nntp-pool.cc pan2.svn-decoder-threads-works!/pan/tasks/nntp-pool.cc
--- pan2.svn-orig/pan/tasks/nntp-pool.cc	2007-03-20 06:17:58.000000000 -0400
+++ pan2.svn-decoder-threads-works!/pan/tasks/nntp-pool.cc	2007-03-18 23:48:05.000000000 -0400
@@ -24,6 +24,7 @@
 #include <pan/general/foreach.h>
 #include <pan/general/log.h>
 #include "nntp-pool.h"
+#include "globals.h"
 
 using namespace pan;
 
@@ -37,11 +38,9 @@
 }
 
 NNTP_Pool :: NNTP_Pool (const Quark        & server,
-                        ServerInfo         & server_info,
-                        Socket::Creator    * creator):
+                        ServerInfo         & server_info):
   _server_info (server_info),
   _server (server),
-  _socket_creator (creator),
   _pending_connections (0),
   _active_count (0),
   _time_to_allow_new_connections (0)
@@ -269,7 +268,7 @@
     if (_server_info.get_server_addr (_server, address, port))
     {
       ++_pending_connections;
-      _socket_creator->create_socket (address, port, this);
+      globals::socketCreator().create_socket (address, port, this);
     }
   }
 }
diff --exclude-from=diff-excludes -urN pan2.svn-orig/pan/tasks/nntp-pool.h pan2.svn-decoder-threads-works!/pan/tasks/nntp-pool.h
--- pan2.svn-orig/pan/tasks/nntp-pool.h	2007-03-20 06:17:57.000000000 -0400
+++ pan2.svn-decoder-threads-works!/pan/tasks/nntp-pool.h	2007-03-18 18:19:20.000000000 -0400
@@ -42,8 +42,7 @@
     public:
 
       NNTP_Pool (const Quark       & server,
-                 ServerInfo        & server_info,
-                 Socket::Creator   *);
+                 ServerInfo        & server_info);
 
       virtual ~NNTP_Pool ();
 
@@ -90,7 +89,6 @@
 
       ServerInfo& _server_info;
       const Quark _server;
-      Socket::Creator * _socket_creator;
       int _pending_connections;
 
       struct PoolItem {
diff --exclude-from=diff-excludes -urN pan2.svn-orig/pan/tasks/queue.cc pan2.svn-decoder-threads-works!/pan/tasks/queue.cc
--- pan2.svn-orig/pan/tasks/queue.cc	2007-03-20 06:17:57.000000000 -0400
+++ pan2.svn-decoder-threads-works!/pan/tasks/queue.cc	2007-03-20 13:17:41.000000000 -0400
@@ -33,11 +33,9 @@
 
 Queue :: Queue (ServerInfo         & server_info,
                 TaskArchive        & archive,
-                Socket::Creator    * socket_creator, 
                 bool                 online):
   _server_info (server_info),
   _is_online (online),
-  _socket_creator (socket_creator),
   _needs_saving (false),
   _last_time_saved (0),
   _archive (archive)
@@ -75,6 +73,12 @@
   fire_queue_error (message);
 }
 
+void Queue :: on_progress_error(Progress &prog, const StringView &message)
+{
+  // probably a task decode error
+  fire_queue_error(message);
+}
+
 NNTP_Pool&
 Queue :: get_pool (const Quark& servername)
 {
@@ -87,7 +91,7 @@
   }
   else // have to build one
   {
-    pool = new NNTP_Pool (servername, _server_info, _socket_creator);
+    pool = new NNTP_Pool (servername, _server_info);
     pool->add_listener (this);
     _pools[servername] = pool;
   }
@@ -135,10 +139,20 @@
 
   // upkeep on running tasks... this lets us pop open
   // extra connections if the task can handle >1 connection
-  std::set<Task*> active; 
+  std::set<Task*> active_or_decod; 
   foreach (nntp_to_task_t, _nntp_to_task, it)
-    active.insert (it->second);
-  foreach (std::set<Task*>, active, it)
+    active_or_decod.insert (it->second);
+  // also manage decoding tasks here
+  // upkeep on decodeing tasks.. this has the potential to move tasks
+  // from the 'need_decode' state to the decoding state
+  foreach_const (TaskSet, _tasks, it) { // need to use _tasks, as tmp above might have deleted entries
+    Task * task = *it;
+    const Task::State & state (task->get_state());
+    if (state._work==Task::DECODING || state._work == Task::NEED_DECODE)
+      active_or_decod.insert(task);
+  }
+  // process active or decoding-related tasks..
+  foreach (std::set<Task*>, active_or_decod, it)
     process_task (*it);
 
   // idle socket upkeep
@@ -183,6 +197,9 @@
   std::set<Task*> active_tasks;
   foreach_const (nntp_to_task_t, _nntp_to_task, it)
     active_tasks.insert (it->second);
+  foreach_const (TaskSet, _tasks, it)
+    if ((*it)->get_state()._work == Task::DECODING)  // decoding is considered to be active?
+      active_tasks.insert(*it);
   active = active_tasks.size ();
   total = _tasks.size ();
 }
@@ -191,7 +208,7 @@
 void
 Queue :: give_task_a_connection (Task * task, NNTP * nntp)
 {
-  const bool was_active (task_is_active (task));
+  const bool was_active (task_was_active (task));
   _nntp_to_task[nntp] = task; // it's active now...
   if (!was_active)
     fire_task_active_changed (task, true);
@@ -202,7 +219,7 @@
 
 void
 Queue :: process_task (Task * task)
-{
+{  
   pan_return_if_fail (task!=0);
 
   debug ("in process_task with a task of type " << task->get_type());
@@ -222,7 +239,9 @@
   else if (_stopped.count(task))
   {
     debug ("stopped");
-    // do nothing
+    if (state._work == Task::DECODING) 
+      // notify decoder thread of stop request
+      task->decode_cancel();
   }
   else if (state._health == COMMAND_FAILED)
   {
@@ -234,6 +253,16 @@
     debug ("working");
     // do nothing
   }
+  else if (state._work == Task::DECODING) {
+    debug ("decoding");
+    // do nothing
+  } 
+  else if (state._work == Task::NEED_DECODE) {
+    debug ("need decode");
+    // try to save..
+    task->try_decode();
+      
+  }
   else while (state._work == Task::NEED_NNTP)
   {
     // make the requests...
@@ -251,6 +280,9 @@
 
     give_task_a_connection (task, nntp);
   }
+  const bool is_active = task_is_active(task), was_active = task_was_active(task);
+  if (is_active != was_active) // notify listeners that active may have changed
+      fire_task_active_changed (task, is_active);
 }
 
 /***
@@ -384,6 +416,7 @@
 void
 Queue :: fire_task_active_changed (Task * task, bool active)
 {
+  _activemap[task] = active;
   for (lit it(_listeners.begin()), end(_listeners.end()); it!=end; )
     (*it++)->on_queue_task_active_changed (*this, *task, active);
 }
@@ -434,8 +467,11 @@
 {
   foreach_const (tasks_t, tasks, it) {
     TaskArticle * ta (dynamic_cast<TaskArticle*>(*it));
-    if (ta)
+    if (ta) 
       _mids.insert (ta->get_article().message_id);
+    
+    (*it)->add_listener(this); // so we pick up decode finished events..
+    
   }
 
   if (mode == TOP)
@@ -468,13 +504,26 @@
 }
 
 bool
+Queue :: task_was_active(const Task *task) const
+{
+  activemap_t::const_iterator it = _activemap.find(task);
+  return it == _activemap.end() ? false : it->second;
+}
+
+bool
 Queue :: task_is_active (const Task * task) const
 {
   bool task_has_nntp (false);
   foreach_const (nntp_to_task_t, _nntp_to_task, it)
     if ((task_has_nntp = task==it->second))
       break;
-  return task_has_nntp;
+  return task_has_nntp || task_is_decoding(task);
+}
+
+bool
+Queue :: task_is_decoding (const Task * task) const
+{
+  return task->get_state()._work == Task::DECODING;
 }
 
 void
@@ -490,9 +539,10 @@
   const int index (_tasks.index_of (task));
   pan_return_if_fail (index != -1);
 
-  if (task_is_active (task)) // wait for the NNTPs to finish
+  if (task_is_active (task)) // wait for the NNTPs or decode threads to finish
   {
-    debug ("can't delete this task right now because it's got server connections");
+    debug ("can't delete this task right now because it's got server connections or is decoding");
+    if (task_is_decoding(task)) task->decode_cancel();
     _removing.insert (task);
   }
   else // no NNTPs working, we can remove right now.
@@ -504,6 +554,8 @@
     _stopped.erase (task);
     _removing.erase (task);
     _tasks.remove (index);
+    if (_activemap[task]) fire_task_active_changed(task, false); // this is needed because sometimes the task is deleted and process_task may erroneously fire task_active_changed after this point...?
+    _activemap.erase(task);
     delete task;
   }
 
@@ -533,6 +585,20 @@
   foreach (nntp_to_task_t, _nntp_to_task, it) tmp.insert (it->second);
   running.clear ();
   running.insert (running.end(), tmp.begin(), tmp.end());
+
+  tmp.clear();
+  std::vector<Task *> & decoding (setme._decoding.get_container());
+  decoding.clear();
+  foreach (tasks_t, _tasks, it) if ((*it)->get_state()._work == Task::DECODING)
+    tmp.insert (*it);
+  decoding.insert(decoding.begin(), tmp.begin(), tmp.end());
+
+  tmp.clear();
+  std::vector<Task *> & decoding_q (setme._decoding_q.get_container());
+  decoding_q.clear();
+  foreach (tasks_t, _tasks, it) if ((*it)->get_state()._work == Task::NEED_DECODE)
+    tmp.insert (*it);
+  decoding_q.insert(decoding_q.begin(), tmp.begin(), tmp.end());
 }
 
 void
@@ -562,8 +628,8 @@
     _nntp_to_task.erase (nntp);
 
     // take care of the task's state
-    const bool is_active (task_is_active (task));
-    if (!is_active) // if it's not active anymore...
+    const bool is_active (task_is_active (task)), was_active(task_was_active(task));
+    if (is_active != was_active) // if it's not active anymore...
       fire_task_active_changed (task, is_active);
 
     // return the nntp to the pool
@@ -676,3 +742,9 @@
   setme_KiBps = KiBps;
   setme_connections = connections;
 }
+
+void
+Queue :: on_progress_finished(Progress &prog, int status)
+{  
+  upkeep(); // call upkeep here to potentially give other tasks in the NEED_DECODE state the opportunity to start a decode going
+}
diff --exclude-from=diff-excludes -urN pan2.svn-orig/pan/tasks/queue.h pan2.svn-decoder-threads-works!/pan/tasks/queue.h
--- pan2.svn-orig/pan/tasks/queue.h	2007-03-20 06:17:58.000000000 -0400
+++ pan2.svn-decoder-threads-works!/pan/tasks/queue.h	2007-03-20 13:17:41.000000000 -0400
@@ -57,10 +57,11 @@
   class Queue:
     public NNTP::Source,
     private NNTP_Pool::Listener,
-    private AdaptableSet<Task*, TaskWeakOrdering>::Listener
+    private AdaptableSet<Task*, TaskWeakOrdering>::Listener,
+    private Progress::Listener
   {
     public:
-      Queue (ServerInfo&, TaskArchive&, Socket::Creator*, bool online);
+      Queue (ServerInfo&, TaskArchive&, bool online);
       virtual ~Queue ();
 
       typedef std::vector<Task*> tasks_t;
@@ -95,7 +96,7 @@
       void get_full_connection_counts (std::vector<ServerConnectionCounts>& setme) const;
 
     public:
-      enum TaskState { RUNNING, QUEUED, STOPPED, REMOVING };
+      enum TaskState { RUNNING, QUEUED, STOPPED, REMOVING, DECODING, QUEUED_FOR_DECODE };
 
       /**
        * An ordered collection of tasks and their corresponding TaskState s.
@@ -108,12 +109,16 @@
           sorted_tasks_t _stopped;
           sorted_tasks_t _running;
           sorted_tasks_t _removing;
+          sorted_tasks_t _decoding;
+          sorted_tasks_t _decoding_q;
         public:
           tasks_t tasks;
           TaskState get_state (Task* task) const {
             if (_removing.count(task)) return REMOVING;
             if (_stopped.count(task)) return STOPPED;
             if (_running.count(task)) return RUNNING;
+            if (_decoding.count(task)) return DECODING;
+            if (_decoding_q.count(task)) return QUEUED_FOR_DECODE;
             if (_queued.count(task)) return QUEUED;
             return STOPPED;
           }
@@ -147,6 +152,10 @@
       virtual void on_pool_has_nntp_available (const Quark& server);
       virtual void on_pool_error (const Quark& server, const StringView& message);
 
+    private: // inherited from Progress::Listener
+      void on_progress_finished(Progress &, int status);
+      void on_progress_error(Progress &, const StringView &msg);
+
     protected:
       void process_task (Task *);
       void give_task_a_connection (Task*, NNTP*);
@@ -155,13 +164,16 @@
       Task* find_first_task_needing_server (const Quark& server);
       bool find_best_server (const Task::State::unique_servers_t& servers, Quark& setme);
       bool task_is_active (const Task*) const;
+      bool task_is_decoding (const Task*) const;
+      bool task_was_active (const Task*) const;
 
       typedef std::map<NNTP*,Task*> nntp_to_task_t;
       nntp_to_task_t _nntp_to_task;
 
       std::set<Task*> _removing;
       std::set<Task*> _stopped;
-      Socket::Creator * _socket_creator;
+      typedef std::map<const Task *, bool> activemap_t;
+      activemap_t _activemap;
 
     protected:
       virtual void fire_tasks_added  (int index, int count);
diff --exclude-from=diff-excludes -urN pan2.svn-orig/pan/tasks/socket-impl-gio.cc pan2.svn-decoder-threads-works!/pan/tasks/socket-impl-gio.cc
--- pan2.svn-orig/pan/tasks/socket-impl-gio.cc	2007-03-20 06:17:57.000000000 -0400
+++ pan2.svn-decoder-threads-works!/pan/tasks/socket-impl-gio.cc	2007-03-20 02:29:46.000000000 -0400
@@ -38,6 +38,7 @@
 
 #include <pan/general/file-util.h>
 #include <pan/general/log.h>
+#include "globals.h"
 
 #ifdef G_OS_WIN32
   // this #define is necessary for mingw
@@ -383,7 +384,7 @@
       g_string_prepend_len (g, _partial_read.c_str(), _partial_read.size());
       _partial_read.clear ();
 
-      debug ("read [" << g->str << "]");
+      debug_v ("read [" << g->str << "]"); // verbose debug, if --debug --debug was on the command-line
       increment_xfer_byte_count (g->len);
       if (g_str_has_suffix (g->str, "\r\n"))
         g_string_truncate (g, g->len-2);
@@ -554,7 +555,8 @@
 
 namespace
 {
-  struct ThreadInfo
+  struct ThreadWorker : public WorkerPool::Worker,
+                        public WorkerPool::Listener
   {
     std::string host;
     int port;
@@ -564,27 +566,29 @@
     Socket * socket;
     std::string err;
 
-    ThreadInfo (const StringView& h, int p, Socket::Creator::Listener *l):
+    ThreadWorker (const StringView& h, int p, Socket::Creator::Listener *l):
       host(h), port(p), listener(l), ok(false), socket(0) {}
+
+    void do_work(void *ignored); // called in thread, initiate connections in thread to avoid blocking pan when connections fail or are slow
+    void on_work_complete(void *ignored); // called in main thread when done, passes results to rest of app
   };
 
-  gboolean socket_created_idle (gpointer info_gpointer)
+
+  void ThreadWorker::on_work_complete (void *ignored)
   {
-    ThreadInfo * info (static_cast<ThreadInfo*>(info_gpointer));
-    if (!info->err.empty())
-      Log :: add_err (info->err.c_str());
-    info->listener->on_socket_created (info->host, info->port, info->ok, info->socket);
-    delete info;
-    return false;
+    // pass results to main thread...
+    (void) ignored;
+    if (!err.empty())   Log :: add_err (err.c_str());
+    listener->on_socket_created (host, port, ok, socket);
+    // NB: WorkerPool framework auto-deletes us after returning from this function
   }
 
-  void create_socket_thread_func (gpointer info_gpointer, gpointer unused)
+  void ThreadWorker::do_work (void *ignored)
   {
+    (void)ignored;
     //std::cerr << LINE_ID << " creating a socket in worker thread...\n";
-    ThreadInfo * info (static_cast<ThreadInfo*>(info_gpointer));
-    info->socket = new GIOChannelSocket ();
-    info->ok = info->socket->open (info->host, info->port, info->err);
-    g_idle_add (socket_created_idle, info); // pass results to main thread...
+    socket = new GIOChannelSocket ();
+    ok = socket->open (host, port, err);    
   }
 }
   
@@ -595,13 +599,6 @@
 {
   ensure_module_inited ();
 
-  static GThreadPool * pool (0);
-  if (!pool)
-    pool = g_thread_pool_new (create_socket_thread_func, 0, 4, true, 0);
-
-  // farm this out to a worker thread so that the main thread
-  // doesn't block while we open the socket.
-  g_thread_pool_push (pool, 
-                      new ThreadInfo (host, port, listener),
-                      NULL);
+  ThreadWorker *w = new ThreadWorker(host, port, listener);
+  globals::workerPool().push_work(w, 0, w, true);
 }
diff --exclude-from=diff-excludes -urN pan2.svn-orig/pan/tasks/task-article-decoder.cc pan2.svn-decoder-threads-works!/pan/tasks/task-article-decoder.cc
--- pan2.svn-orig/pan/tasks/task-article-decoder.cc	1969-12-31 19:00:00.000000000 -0500
+++ pan2.svn-decoder-threads-works!/pan/tasks/task-article-decoder.cc	2007-03-21 19:18:05.000000000 -0400
@@ -0,0 +1,354 @@
+/* -*- Mode: C++; tab-width: 2; indent-tabs-mode: nil; c-basic-offset: 2 -*- */
+/*
+ * Pan - A Newsreader for Gtk+
+ * Copyright (C) 2007  Calin Culianu <calin@ajvar.org>
+ *
+ * This program is free software; you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation; version 2 of the License.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program; if not, write to the Free Software
+ * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
+ */
+#include <config.h>
+#include <algorithm>
+#include <cassert>
+#include <cerrno>
+#include <ostream>
+#include <sstream>
+#include <list>
+extern "C" {
+#  define PROTOTYPES
+#  include <uulib/uudeview.h>
+#  include <glib/gi18n.h>
+};
+#include <pan/general/debug.h>
+#include <pan/general/file-util.h>
+#include <pan/general/foreach.h>
+#include <pan/general/locking.h>
+#include <pan/general/log.h>
+#include <pan/usenet-utils/mime-utils.h>
+#include <pan/data/article-cache.h>
+#include "task-article-decoder.h"
+
+using namespace pan;
+
+//static
+bool TaskArticle::Decoder::checked_out = false;
+
+/* static */
+TaskArticle::Decoder *
+TaskArticle::Decoder::getInstance()
+{
+  static Decoder *instance = 0; // ptr to instance is here just to avoid unguarded access
+
+  if (!instance) instance = new Decoder;
+  return instance;
+}
+
+/* static */
+TaskArticle::Decoder *
+TaskArticle::Decoder::check_out()
+{
+  if (!checked_out) {
+    checked_out = true;
+    return getInstance();
+  }
+  return 0;
+}
+
+/* static */
+void 
+TaskArticle::Decoder::check_in(Decoder *d)
+{
+  (void)d;
+  assert (d == getInstance());
+  assert (checked_out);
+  checked_out = false;
+}
+
+// re-initialize the object which gets re-used a lot
+void 
+TaskArticle :: Decoder :: init ( TaskArticle *t,
+                                 const Quark & sp,
+                                 const ArticleCache::strings_t &f,
+                                 const TaskArticle::SaveMode & sm 
+                                )
+{  
+  disable_progress_update();
+  task = t;
+  save_path = sp.to_string();
+  filenames = f;
+  save_mode = sm;
+  mark_read = false;
+  log_urgents.clear();
+  log_errors.clear();
+  log_infos.clear();
+  base_percent = percent = 0;
+  please_stop = false; // clear WorkerPool::worker state...
+}
+
+TaskArticle :: Decoder :: Decoder()
+{
+  gsourceid = -1;
+}
+
+TaskArticle :: Decoder :: ~Decoder()
+{
+  disable_progress_update();
+}
+
+void 
+TaskArticle :: Decoder :: do_work(void *ignored) // save article in another thread to avoid network stalls
+{
+  static const int bufsz = 4096;
+  char buf[bufsz];
+
+  (void)ignored;
+  enable_progress_update();
+
+  // NOTE THIS WHOLE METHOD RUNS IN ONE SEPARATE THREAD -- so be sure to keep that in mind..
+  if (save_mode & TaskArticle::RAW)
+  {   
+    int i = 0;
+    foreach_const (ArticleCache::strings_t, filenames, it)
+    {
+      if (please_stop) break; // poll WorkerPool::Worker stop flag
+
+      gchar * contents (0);
+      gsize length (0);
+      if (g_file_get_contents (it->c_str(), &contents, &length, NULL) && length>0)
+      {
+        file :: ensure_dir_exists (save_path.c_str());
+        gchar * basename (g_path_get_basename (it->c_str()));
+        gchar * filename (g_build_filename (save_path.c_str(), basename, NULL));
+        FILE * fp = fopen (filename, "w+");
+
+        mut.lock();
+        current_file = filename; // save the filename in class so that progress code can see it potentially
+        mut.unlock();
+
+        if (!fp) {
+          g_snprintf(buf, bufsz, _("Couldn't save file \"%s\": %s"), filename, file::pan_strerror(errno));
+          log_errors.push_back (buf); // log error
+        } else {
+          fwrite (contents, 1, (size_t)length, fp);
+          fclose (fp);
+        }
+        g_free (filename);
+        g_free (basename);
+      }
+      g_free (contents);
+      
+      // update percent for progress_update_timer_func
+      mut.lock();
+      percent = ++i*100/filenames.size();
+      mut.unlock();
+    }
+  }
+
+  if (save_mode & TaskArticle::DECODE)
+  {
+    // decode
+    int res, step = 0;
+    if (((res = UUInitialize())) != UURET_OK)
+      log_errors.push_back(_("Error initializing uulib")); // log error
+    else
+    {
+      UUSetMsgCallback (this, uu_log);
+      UUSetOption (UUOPT_DESPERATE, 1, NULL); // keep incompletes -- they're still useful to par2
+
+      int i (0);
+      foreach_const (ArticleCache::strings_t, filenames, it) {
+        if (please_stop) break; // poll WorkerPool::Worker stop flag
+        if ((res = UULoadFileWithPartNo (const_cast<char*>(it->c_str()), 0, 0, ++i)) != UURET_OK) {
+          g_snprintf(buf, bufsz, 
+                   _("Error reading from %s: %s"), 
+                   it->c_str(),
+                   (res==UURET_IOERR) 
+                   ?  file::pan_strerror (UUGetOption (UUOPT_ERRNO, NULL, 
+                                                       NULL, 0)) 
+                   : UUstrerror(res));
+          log_errors.push_back(buf); // log error
+        }
+        mut.lock();
+        int tmp = percent = base_percent =  ++step*10/filenames.size(); // update percentage progress member .. not this goes up to 10% and the remaining 90% is calculated from the actually uuprogress given us by uulib
+        mut.unlock();
+        debug("uudecoder thread: first pass progress " << tmp << "%");
+      }
+
+
+      i = 0;
+      uulist * item;
+
+      UUSetBusyCallback (this, uu_busy_poll, 1000); // 1.0 secs busy poll?
+
+      i = 0;
+      while ((item = UUGetFileListItem (i++)))
+      {
+        if (please_stop) break; // poll WorkerPool::Worker stop flag
+
+        mut.lock();
+        ++itemno; // save it in class
+        mut.unlock();
+
+        // make sure the directory exists...
+        if (!save_path.empty())
+          file :: ensure_dir_exists (save_path.c_str());
+
+        // find a unique filename...
+        char * fname (0);
+        for (int i=0; ; ++i) {
+          std::string basename ((item->filename && *item->filename)
+                                ? item->filename
+                                : "pan-saved-file");
+          if (i) {
+            g_snprintf (buf, bufsz, "_copy_%d", i+1); // we don't want "_copy_1"
+            // try to preserve any extension
+            std::string::size_type dotwhere = basename.find_last_of(".");
+            if (dotwhere != basename.npos) {// if we found a dot
+          	  std::string bn (basename, 0, dotwhere); // everything before the last dot
+          	  std::string sf (basename, dotwhere, basename.npos); // the rest
+          	  // add in a substring to make it unique and enable things like "rm -f *_copy_*"
+          	  basename = bn + buf + sf;
+            }else{
+          	  basename += buf;
+            }
+          }
+          fname = save_path.empty()
+            ? g_strdup (basename.c_str())
+            : g_build_filename (save_path.c_str(), basename.c_str(), NULL);
+          if (!file::file_exists (fname))
+            break;
+          g_free (fname);
+        }
+
+        // decode the file...
+        if ((res = UUDecodeFile (item, fname)) == UURET_OK) {
+          g_snprintf(buf, bufsz,_("Saved \"%s\""), fname);
+          log_infos.push_back(buf); // log info
+        } else if (res == UURET_NODATA) {
+          // silently let this error by... user probably tried to
+          // save attachements on a text-only post
+        } else {
+          const int the_errno (UUGetOption (UUOPT_ERRNO, NULL, NULL, 0));
+          if (res==UURET_IOERR && the_errno==ENOSPC) {
+            g_snprintf (buf, bufsz, _("Error saving \"%s\":\n%s. %s"), fname, file::pan_strerror(the_errno), "ENOSPC");
+            log_urgents.push_back(buf); // log this to the urgent log
+          } else {
+            g_snprintf (buf, bufsz,_("Error saving \"%s\":\n%s."),
+                             fname,
+                             res==UURET_IOERR ? file::pan_strerror(the_errno) : UUstrerror(res));
+            log_errors.push_back(buf); // log error
+          }
+        }
+
+        // cleanup
+        g_free (fname);
+
+      }
+
+      mark_read = true;
+    }
+    UUCleanUp ();
+  }
+
+  if (please_stop) 
+    debug("got notification to stop early, decode might not be finished..");
+  disable_progress_update();
+}
+
+/* static */
+void 
+TaskArticle :: Decoder :: uu_log (void* thiz, char* message, int severity)
+{
+    Decoder *self = reinterpret_cast<Decoder *>(thiz);
+    char * pch (g_locale_to_utf8 (message, -1, 0, 0, 0));
+
+    if (severity == UUMSG_PANIC || severity==UUMSG_FATAL || severity==UUMSG_ERROR)
+      self->log_errors.push_back (pch ? pch : message);
+    else if (severity == UUMSG_WARNING || severity==UUMSG_NOTE)
+      self->log_infos.push_back (pch ? pch : message);
+
+    g_free (pch);
+}
+
+/* static */
+int
+TaskArticle :: Decoder :: uu_busy_poll(void *data, uuprogress *p)
+{
+  Decoder *thiz = reinterpret_cast<Decoder *>(data);
+  if (thiz->please_stop) {
+    debug("uudecoder thread: got stop request, aborting early");
+    return 1;
+  }
+
+  debug("uudecoder thread: uuprogress is " << p->percent << " percent on " << p->fsize << "b file `" << p->curfile << "' part " << p->partno << " of " << p->numparts);
+  thiz->mut.lock();
+  double pct = p->percent, nitems = p->numparts, basepct = thiz->base_percent, item = p->partno;
+  if (p->numparts == 1) // we are in phase II of uudecode.. so pick up percent here by offsetting from remembered base_percent value computed in previous phases
+    thiz->percent =  int( pct = (pct/100.0) * (100.0-basepct) + basepct ); // complicated way to update percentage -- this is because the uudecode progress is reset to 0 for each phase so we have to fudge its value
+  else { // phase 1 takes about 10% of total time
+    thiz->percent = thiz->base_percent = int(pct = 10.0 + p->partno*10/p->numparts);
+  }
+  thiz->current_file = p->curfile;
+  thiz->mut.unlock();
+  debug("uudecoder thread: calculated percent is " << pct);
+
+  return 0;
+}
+
+void 
+TaskArticle::Decoder::enqueue_work_in_thread(TaskArticle *listener,
+                                             void *listener_data,
+                                             const Quark & save_path,
+                                             const ArticleCache::strings_t & filenames,
+                                             const TaskArticle::SaveMode & save_mode)
+{
+  init(listener, save_path, filenames,  save_mode);
+
+  // gentlemen, start your saving...
+  globals::workerPool().push_work(this,/* who is the worker? */
+                                  listener_data,/* void* data passed listener*/
+                                  listener,   /* who is the listener? */
+                                  false   /* don't auto-delete worker */);
+}
+
+gboolean 
+TaskArticle::Decoder::progress_update_timer_func(gpointer decoder)
+{
+  Decoder *thiz = reinterpret_cast<Decoder *>(decoder);
+  Task *task = thiz->task;
+  if (!task || thiz->was_cancelled()) return false;
+  
+  thiz->mut.lock();
+  int percent = thiz->percent;
+  std::string f = thiz->current_file;
+  thiz->mut.unlock();
+  task->set_step(percent);
+  task->set_status_va(_("Decoding %s"), f.c_str());
+
+  debug("setting task progress to: " << percent << "% file: " << f);
+
+  return true; // keep timer func running
+}
+
+void TaskArticle::Decoder::enable_progress_update()
+{
+  disable_progress_update(); // disable any running gsources, if any
+  gsourceid = g_timeout_add(1000, progress_update_timer_func, this);
+}
+
+void TaskArticle::Decoder::disable_progress_update()
+{
+  if (gsourceid > -1) {
+    g_source_remove (gsourceid);
+    gsourceid = -1;
+  }
+}
diff --exclude-from=diff-excludes -urN pan2.svn-orig/pan/tasks/task-article-decoder.h pan2.svn-decoder-threads-works!/pan/tasks/task-article-decoder.h
--- pan2.svn-orig/pan/tasks/task-article-decoder.h	1969-12-31 19:00:00.000000000 -0500
+++ pan2.svn-decoder-threads-works!/pan/tasks/task-article-decoder.h	2007-03-21 19:18:05.000000000 -0400
@@ -0,0 +1,107 @@
+/* -*- Mode: C++; tab-width: 2; indent-tabs-mode: nil; c-basic-offset: 2 -*- */
+/*
+ * Pan - A Newsreader for Gtk+
+ * Copyright (C) 2007  Calin Culianu <calin@ajvar.org>
+ *
+ * This program is free software; you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation; version 2 of the License.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program; if not, write to the Free Software
+ * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
+ */
+#ifndef task_article_decoder_H
+#define task_article_decoder_H
+
+#include <pan/general/debug.h>
+#include <pan/general/file-util.h>
+#include <pan/general/foreach.h>
+#include <pan/general/locking.h>
+#include <pan/general/log.h>
+#include <pan/usenet-utils/mime-utils.h>
+#include <pan/data/article-cache.h>
+extern "C" {
+#  define PROTOTYPES
+#  include <uulib/uudeview.h>
+#  include <glib/gi18n.h>
+};
+#include <glib.h>
+#include "task-article.h"
+#include "globals.h"
+
+namespace pan {
+
+  class TaskArticle::Decoder : protected WorkerPool::Worker
+  {
+  public:
+    
+    void enqueue_work_in_thread(TaskArticle *task,
+                                void *listener_data,
+                                const Quark & save_path,
+                                const ArticleCache::strings_t & filenames,
+                                const TaskArticle::SaveMode & save_mode);
+        
+    ~Decoder();
+    
+    
+    std::string save_path;
+    ArticleCache::strings_t filenames;
+    TaskArticle::SaveMode save_mode;    
+    
+    std::list<std::string> log_urgents, log_errors, log_infos; 
+    bool mark_read;
+
+    /* The below values are automagically polled from the main thread 
+       while the decode task is running, and updates are sent to the Task *
+       object this decoder owns. */
+    Mutex mut;
+    volatile int base_percent, percent, itemno, num_items;
+    std::string current_file; // the current file we are decoding, with path
+
+    TaskArticle *task;
+
+    /// may return null if the singleton is already checked out
+    static Decoder *check_out(); 
+    
+    /** check back in the singleton that was previously checked out -- 
+        so that future check_outs will succeed */
+    static void check_in(Decoder *); 
+    
+    /// cancel an already-running decoder -- note the listener is notified with a on_work_cancelled call rather than a on_work_complete call -- from WorkerPool::Worker interface
+    void cancel() {  WorkerPool::Worker::cancel(); }
+
+  protected:
+    /// from WorkerPool::Worker interface, saves article to filesystem
+    void do_work(void *); 
+    /// just clears member vars for another run in a different thread
+    void init(TaskArticle *t,
+              const Quark & save_path,
+              const ArticleCache::strings_t & filenames,
+              const TaskArticle::SaveMode & save_mode);
+    
+  private:
+    static void uu_log(void *thiz, char *message, int severity);
+    static int uu_busy_poll(void *thiz, uuprogress *p);
+    /// updates Progress * object (aka task) about progress of decode step
+    static gboolean progress_update_timer_func(gpointer decoder);
+    static bool checked_out; // only 1 thing can check this out at once
+    
+    // singleton
+    static Decoder *getInstance();
+
+    Decoder(); // singleton
+
+    int gsourceid;
+    void disable_progress_update();
+    void enable_progress_update();
+  };
+  
+}
+
+#endif
diff --exclude-from=diff-excludes -urN pan2.svn-orig/pan/tasks/task-article.cc pan2.svn-decoder-threads-works!/pan/tasks/task-article.cc
--- pan2.svn-orig/pan/tasks/task-article.cc	2007-03-20 06:17:58.000000000 -0400
+++ pan2.svn-decoder-threads-works!/pan/tasks/task-article.cc	2007-03-20 13:17:41.000000000 -0400
@@ -2,6 +2,7 @@
 /*
  * Pan - A Newsreader for Gtk+
  * Copyright (C) 2002-2006  Charles Kerr <charles@rebelbase.com>
+ * Copyright (C) 2007  Calin Culianu <calin@ajvar.org>
  *
  * This program is free software; you can redistribute it and/or modify
  * it under the terms of the GNU General Public License as published by
@@ -23,10 +24,11 @@
 #include <cerrno>
 #include <ostream>
 #include <sstream>
+#include <list>
 extern "C" {
-  #define PROTOTYPES
-  #include <uulib/uudeview.h>
-  #include <glib/gi18n.h>
+#  define PROTOTYPES
+#  include <uulib/uudeview.h>
+#  include <glib/gi18n.h>
 };
 #include <pan/general/debug.h>
 #include <pan/general/file-util.h>
@@ -34,7 +36,9 @@
 #include <pan/general/log.h>
 #include <pan/usenet-utils/mime-utils.h>
 #include <pan/data/article-cache.h>
+#include "globals.h"
 #include "task-article.h"
+#include "task-article-decoder.h"
 
 using namespace pan;
 
@@ -63,7 +67,7 @@
                             const Article             & article,
                             ArticleCache              & cache,
                             ArticleRead               & read,
-                            Task::Listener            * listener,
+                            Progress::Listener        * listener,
                             SaveMode                    save_mode,
                             const Quark               & save_path):
   Task (save_path.empty() ? "BODIES" : "SAVE", get_description (article, !save_path.empty())),
@@ -74,7 +78,8 @@
   _article (article),
   _time_posted (article.time_posted),
   _finished_proc_has_run (false),
-  _save_mode (save_mode)
+  _save_mode (save_mode),
+  _decoder(0)
 {
   cache.reserve (article.get_part_mids());
 
@@ -126,13 +131,14 @@
   else
     set_status_va (_("Saving %s"), article.subject.c_str());
  
+  if (_needed.empty() && all_bytes) {
+     _state.set_need_decode();
+     set_step(0);
+  }
+
   update_work ();
 }
 
-TaskArticle :: ~TaskArticle ()
-{
-  _cache.release (_article.get_part_mids());
-}
 
 void
 TaskArticle :: update_work ()
@@ -161,10 +167,8 @@
     _state.set_need_nntp (servers);
   else if (working)
     _state.set_working ();
-  else {
-    _state.set_completed ();
+  else if (_state._work == COMPLETED) // this was auto-set by saver process
     set_finished (OK);
-  }
 
   if (_state._work == COMPLETED && !_finished_proc_has_run) {
     _finished_proc_has_run = true;
@@ -259,9 +263,15 @@
   // if we got it or still have other options, we're okay
   _state.set_health (health==COMMAND_FAILED && it->xref.empty() ? COMMAND_FAILED : OK);
 
-  if (health==OK)
+  if (health==OK) {
     _needed.erase (it);
-  else {
+    if (_needed.empty() && _save_mode) { // if we just successfully finished with all needed parts and we need to decode the attachment, set status here
+      _state.set_need_decode();
+      set_step(0); // indicate we have yet to complete this phase by resetting progress
+    } else {
+      _state.set_completed();
+    }
+  } else {
     Needed::buf_t tmp;
     it->buf.swap (tmp); // deallocates the space...
     it->nntp = 0;
@@ -271,130 +281,106 @@
   check_in (nntp, health);
 }
 
-namespace
+void
+TaskArticle :: on_work_cancelled(void *data)
 {
-  void uu_log (void* unused, char* message, int severity)
-  {
-    char * pch (g_locale_to_utf8 (message, -1, 0, 0, 0));
+  (void)data;
 
-    if (severity == UUMSG_PANIC || severity==UUMSG_FATAL || severity==UUMSG_ERROR)
-      Log :: add_err (pch ? pch : message);
-    else if (severity == UUMSG_WARNING || severity==UUMSG_NOTE)
-      Log :: add_info (pch ? pch : message);
+  assert(_decoder);
+  if (!_decoder) return;
 
-    g_free (pch);
-  }
+  _state.set_need_decode();
+  set_step(0);
+  Decoder::check_in(_decoder);
+  _decoder = 0;
 }
 
+
+// this is called as per the WorkerPool::Listener interface and it means that the save thread finished
 void
-TaskArticle :: on_finished ()
+TaskArticle :: on_work_complete(void *data) 
 {
-  const Article::mid_sequence_t mids (_article.get_part_mids());
-  const ArticleCache :: strings_t filenames (_cache.get_filenames (mids));
+  (void) data;
 
-  if (_save_mode & RAW)
-  {
-    foreach_const (ArticleCache::strings_t, filenames, it)
-    {
-      gchar * contents (0);
-      gsize length (0);
-      if (g_file_get_contents (it->c_str(), &contents, &length, NULL) && length>0)
-      {
-        file :: ensure_dir_exists (_save_path.c_str());
-        gchar * basename (g_path_get_basename (it->c_str()));
-        gchar * filename (g_build_filename (_save_path.c_str(), basename, NULL));
-        FILE * fp = fopen (filename, "w+");
-        if (!fp)
-          Log::add_err_va (_("Couldn't save file \"%s\": %s"), filename, file::pan_strerror(errno));
-        else {
-          fwrite (contents, 1, (size_t)length, fp);
-          fclose (fp);
-        }
-        g_free (filename);
-        g_free (basename);
-      }
-      g_free (contents);
-    }
-  }
+  assert(_decoder);
+  if (!_decoder) return;
 
-  if (_save_mode & DECODE)
-  {
-    // decode
-    int res;
-    if (((res = UUInitialize())) != UURET_OK)
-      Log::add_err (_("Error initializing uulib"));
-    else
-    {
-      UUSetMsgCallback (NULL, uu_log);
-      UUSetOption (UUOPT_DESPERATE, 1, NULL); // keep incompletes -- they're still useful to par2
+  bool have_errs = false;
 
-      int i (0);
-      foreach_const (ArticleCache::strings_t, filenames, it) {
-        if ((res = UULoadFileWithPartNo (const_cast<char*>(it->c_str()), 0, 0, ++i)) != UURET_OK)
-          Log::add_err_va (_("Error reading from %s: %s"), it->c_str(),
-            (res==UURET_IOERR) ?  file::pan_strerror (UUGetOption (UUOPT_ERRNO, NULL, NULL, 0)) : UUstrerror(res));
-      }
-
-      i = 0;
-      uulist * item;
-      while ((item = UUGetFileListItem (i++)))
-      {
-        // make sure the directory exists...
-        if (!_save_path.empty())
-          file :: ensure_dir_exists (_save_path.c_str());
-
-        // find a unique filename...
-        char * fname (0);
-        for (int i=0; ; ++i) {
-          std::string basename ((item->filename && *item->filename)
-                                ? item->filename
-                                : "pan-saved-file");
-          if (i) {
-            char buf[32];
-            g_snprintf (buf, sizeof(buf), "_copy_%d", i+1); // we don't want "_copy_1"
-            // try to preserve any extension
-            std::string::size_type dotwhere = basename.find_last_of(".");
-            if (dotwhere != basename.npos) {// if we found a dot
-          	  std::string bn (basename, 0, dotwhere); // everything before the last dot
-          	  std::string sf (basename, dotwhere, basename.npos); // the rest
-          	  // add in a substring to make it unique and enable things like "rm -f *_copy_*"
-          	  basename = bn + buf + sf;
-            }else{
-          	  basename += buf;
-            }
-          }
-          fname = _save_path.empty()
-            ? g_strdup (basename.c_str())
-            : g_build_filename (_save_path.c_str(), basename.c_str(), NULL);
-          if (!file::file_exists (fname))
-            break;
-          g_free (fname);
-        }
-
-        // decode the file...
-        if ((res = UUDecodeFile (item, fname)) == UURET_OK) {
-          Log::add_info_va (_("Saved \"%s\""), fname);
-        } else if (res == UURET_NODATA) {
-          // silently let this error by... user probably tried to
-          // save attachements on a text-only post
-        } else {
-          const int the_errno (UUGetOption (UUOPT_ERRNO, NULL, NULL, 0));
-          if (res==UURET_IOERR && the_errno==ENOSPC)
-            Log::add_urgent_va (_("Error saving \"%s\":\n%s. %s"), fname, file::pan_strerror(the_errno), "ENOSPC");
-          else
-            Log::add_err_va (_("Error saving \"%s\":\n%s."),
-                             fname,
-                             res==UURET_IOERR ? file::pan_strerror(the_errno) : UUstrerror(res));
-        }
+  // when we get here, our decoder thread finished.. so pick up the logs it
+  // generated and put them in the log here in the main thread, and also
+  // update the _read object to mark the article read, and also 
+  // notify listeners that task_decode stuff is completed
+  foreach_const(std::list<std::string>, _decoder->log_urgents, it) 
+    Log :: add_urgent(it->c_str()), have_errs = true;
+  foreach_const(std::list<std::string>, _decoder->log_errors, it) 
+    Log :: add_err(it->c_str()), have_errs = true;
+  foreach_const(std::list<std::string>, _decoder->log_infos, it) 
+    Log :: add_info(it->c_str());
+
+  if (_decoder->mark_read) _read.mark_read(_article);
+
+  std::string errstr = "";
+  if (have_errs) {
+    errstr = 
+      !_decoder->log_errors.empty() 
+      ? _decoder->log_errors.front() 
+      : _decoder->log_urgents.front();
+  }
+
+  Decoder::check_in(_decoder);
+  _decoder = 0;
+
+  _state.set_completed(); // mark us COMPLETED here.. 
+  set_step(100); // indicate we're done to interested parties
+  if (have_errs) set_error(errstr);
+  
+  update_work(); // may call set_finished for us..
+  debug("decoder thread done");
+}
 
-        // cleanup
-        g_free (fname);
-      }
+bool TaskArticle :: try_decode()
+{
+  bool ret = false;
+  assert(_state._work == NEED_DECODE);
 
-      _read.mark_read (_article);
+  if (_state._work == NEED_DECODE) {
+    if ( (_decoder = Decoder::check_out()) ) { // is thread free?
+      // yay thread is free
+      const Article::mid_sequence_t mids (_article.get_part_mids());
+      const ArticleCache :: strings_t filenames (_cache.get_filenames (mids));
+
+      init_steps(100);
+      _state.set_decoding();
+      
+      _decoder->enqueue_work_in_thread(this, _decoder, _save_path, filenames, _save_mode);
+
+      set_status_va (_("Decoding %s"), _article.subject.c_str());
+      //increment_step();
+      ret = true;
+      debug("decoder thread was free, enqueued work");
+    } else {
+      debug("decoder thread was busy, we'll try later");
     }
-    UUCleanUp ();
   }
+  return ret;
+}
+
+void
+TaskArticle :: decode_cancel ()
+{
+  if (_state._work == DECODING && _decoder) {
+    debug("Decoder cancellation request -- forwarding to decoder object");
+    _decoder->cancel();
+  }
+    // will set the cancel state when our on_work_cancelled callback finally runs and the decode ends  
+}
+
+void
+TaskArticle :: on_finished ()
+{
+  // noop -- used to uudecode and/or save attachements here, but this was
+  // moved to the Decoder object above.. see task-article-decoder.[ch]* and try_decode() above..
 }
 
 unsigned long
@@ -405,3 +391,11 @@
     bytes += (it->part.bytes - it->buf.size());
   return bytes;
 }
+
+
+TaskArticle :: ~TaskArticle ()
+{
+  decode_cancel(); // cancel running decode, in order to release thread asap
+  _decoder = 0;
+  _cache.release (_article.get_part_mids());
+}
diff --exclude-from=diff-excludes -urN pan2.svn-orig/pan/tasks/task-article.h pan2.svn-decoder-threads-works!/pan/tasks/task-article.h
--- pan2.svn-orig/pan/tasks/task-article.h	2007-03-20 06:17:58.000000000 -0400
+++ pan2.svn-decoder-threads-works!/pan/tasks/task-article.h	2007-03-20 13:17:41.000000000 -0400
@@ -27,14 +27,18 @@
 #include <pan/data/xref.h>
 #include <pan/tasks/nntp.h>
 #include <pan/tasks/task.h>
+#include <pan/tasks/worker-pool.h>
 
 namespace pan
 {
+
   /**
    * Task for downloading, and optionally decoding, articles
    * @ingroup tasks
    */
-  class TaskArticle: public Task, private NNTP::Listener
+  class TaskArticle: public Task, 
+                     private NNTP::Listener, 
+                     private WorkerPool::Listener
   {
     public: // life cycle
 
@@ -45,16 +49,25 @@
                    const Article      & article,
                    ArticleCache       & cache,
                    ArticleRead        & read,
-                   Task::Listener     * l=0,
+                   Progress::Listener* l=0,
                    SaveMode             save_mode = NONE,
                    const Quark        & save_path = Quark());
       virtual ~TaskArticle ();
       time_t get_time_posted () const { return _time_posted; }
       const Quark& get_save_path () const { return _save_path; }
       const Article& get_article () const { return _article; }
-        
+
     public: // Task subclass
       virtual unsigned long get_bytes_remaining () const;
+    
+    /** only call this for tasks in the NEED_DECODE state
+     * attempts to acquire the saver thread and start saving
+     * returns false if failed or true if the save process started
+     * (intended to be used with the Queue class). If true is returned, 
+     * a side-effect is that the task is now in the DECODING state.
+     */    
+      bool try_decode();
+      void decode_cancel();
 
     private: // Task subclass
       virtual void use_nntp (NNTP * nntp);
@@ -63,6 +76,10 @@
       virtual void on_nntp_line  (NNTP*, const StringView&);
       virtual void on_nntp_done  (NNTP*, Health, const StringView&);
 
+    private: // WorkerPool::Listener interface
+      void on_work_complete(void *); 
+      void on_work_cancelled(void *); 
+
     protected:
       const Quark _save_path;
       const ServerRank& _server_rank;
@@ -78,6 +95,10 @@
       const SaveMode _save_mode;
       typedef std::map<Quark,int> stats_t;
       stats_t _stats;
+           
+      class Decoder; // see task-article-decoder.h
+      friend class Decoder;
+      Decoder *_decoder;
 
     private:
       struct Needed {
@@ -93,6 +114,7 @@
       needed_t _needed;
 
       void update_work ();
+
   };
 }
 
diff --exclude-from=diff-excludes -urN pan2.svn-orig/pan/tasks/task.cc pan2.svn-decoder-threads-works!/pan/tasks/task.cc
--- pan2.svn-orig/pan/tasks/task.cc	2007-03-20 06:17:57.000000000 -0400
+++ pan2.svn-decoder-threads-works!/pan/tasks/task.cc	2007-03-20 13:17:41.000000000 -0400
@@ -20,6 +20,7 @@
 #include <config.h>
 #include <pan/general/debug.h>
 #include <pan/general/messages.h>
+#include <pan/general/foreach.h>
 #include "task.h"
 
 using namespace pan;
@@ -75,3 +76,4 @@
    return oldest_time;
 }
 #endif
+
diff --exclude-from=diff-excludes -urN pan2.svn-orig/pan/tasks/task.h pan2.svn-decoder-threads-works!/pan/tasks/task.h
--- pan2.svn-orig/pan/tasks/task.h	2007-03-20 06:17:57.000000000 -0400
+++ pan2.svn-decoder-threads-works!/pan/tasks/task.h	2007-03-20 13:17:41.000000000 -0400
@@ -51,7 +51,11 @@
             /** Task is waiting on an nntp connection */
             NEED_NNTP,
             /** Task is running */
-            WORKING
+            WORKING,
+            /** Task waiting for decoder thread */
+            NEED_DECODE,
+            /** Task saving */
+            DECODING,
          };
 
          /**
@@ -85,6 +89,12 @@
                void set_need_nntp (const Quark& server) {
                   _work=NEED_NNTP; _servers.clear(); _servers.insert(server); }
 
+               void set_need_decode () {
+                   _work = NEED_DECODE; _servers.clear(); }
+
+               void set_decoding () {
+                   _work = DECODING; _servers.clear(); }
+
                void set_health (Health h) {
                   _health = h; }
 
@@ -108,6 +118,11 @@
          const Quark& get_type () const { return _type; }
 
          virtual unsigned long get_bytes_remaining () const = 0;
+          
+         /// see TaskArticle for a class that actually uses this method
+         virtual bool try_decode() { return false; }
+         /// cancel an already-running decode -- TaskArticle actually uses this
+         virtual void decode_cancel() { } 
 
       protected:
 
diff --exclude-from=diff-excludes -urN pan2.svn-orig/pan/tasks/worker-pool.cc pan2.svn-decoder-threads-works!/pan/tasks/worker-pool.cc
--- pan2.svn-orig/pan/tasks/worker-pool.cc	1969-12-31 19:00:00.000000000 -0500
+++ pan2.svn-decoder-threads-works!/pan/tasks/worker-pool.cc	2007-03-21 19:18:05.000000000 -0400
@@ -0,0 +1,150 @@
+/* -*- Mode: C++; tab-width: 2; indent-tabs-mode: nil; c-basic-offset: 2 -*- */
+/*
+ * Pan - A Newsreader for Gtk+
+ * Copyright (C) 2007  Calin Culianu <calin@ajvar.org>
+ *
+ * This program is free software; you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation; version 2 of the License.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program; if not, write to the Free Software
+ * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
+ */
+#include <pan/general/debug.h>
+#include <pan/general/foreach.h>
+#include "worker-pool.h"
+#include <cassert>
+#include <set>
+#include <map>
+
+namespace pan
+{
+  
+  /* static */ 
+  WorkerPool::WorkerSet & 
+  WorkerPool::all_workers() 
+  {
+    static WorkerPool::WorkerSet *all_workers_ptr = 0;
+
+    if (!all_workers_ptr) 
+      all_workers_ptr = new WorkerSet;
+    return *all_workers_ptr;
+  }
+
+  WorkerPool::Worker::Worker() : please_stop(false) 
+  {
+    all_workers().insert(this);
+  }
+
+  WorkerPool::Worker::~Worker() 
+  {
+    all_workers().erase(this);
+  }
+
+  void
+  WorkerPool::cancelAllWorkers()
+  {
+    debug("notifying all workers to stop");
+    foreach(WorkerSet, all_workers(), it)
+      (*it)->cancel();
+  }
+
+  struct WorkerPool::Work 
+  {
+    Work() : worker(0), listener(0), delete_worker(false), data(0) {}
+    Worker *worker;
+    WorkerPool *pool;
+    Listener *listener;
+    bool delete_worker;
+    void *data;
+  };
+
+  WorkerPool::WorkerPool(int nthr, bool exclusive)
+  {
+    if (!g_thread_supported ()) g_thread_init (NULL);
+    //g_thread_pool_set_max_idle_time(1000); //1s idle time enough to reap idle threads? if users of this class care, perhaps make this a setable global property..
+    assert(!exclusive || nthr > -1); // assert that for exclusive threads, nthr cannot be negative
+    tpool = g_thread_pool_new(thr_wrapper, 
+                              reinterpret_cast<gpointer>(this),
+                              nthr,
+                              exclusive, /* exclusive? */
+                              0);
+  }
+  
+  WorkerPool::~WorkerPool()
+  {    
+    debug("Deleting thread pool 0x" << ((void *)this) << " with max threads: " << maxThreads() << " num threads: " << numThreads() << " unprocessed: " << unprocessed());
+
+    // ask my_workers to die
+    foreach(WorkerSet, my_workers, it)
+      (*it)->cancel();
+
+    g_thread_pool_free(tpool, false, true);
+  }
+
+  /* static */ 
+  void WorkerPool::thr_wrapper(gpointer data, gpointer user_data)
+  {
+    WorkerPool *self = reinterpret_cast<WorkerPool *>(user_data);
+    Work *work = reinterpret_cast<Work *>(data);
+    self->doWork(work); // call class member
+  }
+
+  /* static */ 
+  gboolean WorkerPool::finalize(gpointer data)
+  {
+    Work *work = reinterpret_cast<Work *>(data);
+    if (work->listener) {
+      if (work->worker->was_cancelled())
+        work->listener->on_work_cancelled(work->data);
+      else
+        work->listener->on_work_complete(work->data);
+    }
+    work->pool->my_workers.erase(work->worker);
+    if (work->delete_worker) delete work->worker;
+    delete work;
+    return false; // tell main loop not to call us again
+  }
+
+
+  void WorkerPool::doWork(Work *work) 
+  {
+    /* do work here .. */
+    work->worker->do_work(work->data);
+    g_idle_add(finalize, work); /* deletes work, also may call listener funcs, etc */
+  }
+
+  void WorkerPool::push_work(Worker *worker, void * data, Listener *listener, bool delworker) 
+  {
+    Work *w = new Work;
+    w->worker = worker;
+    w->listener = listener;
+    w->delete_worker = delworker;
+    w->data = data;
+    w->pool = this;
+    my_workers.insert(worker);
+    g_thread_pool_push(tpool, w, NULL); // ends up invokint thr_wrapper in thread    
+  }
+
+
+  void WorkerPool::setMaxThreads(int num) {
+    g_thread_pool_set_max_threads(tpool, num, NULL);
+  }
+  int WorkerPool::maxThreads(void) const {
+    return g_thread_pool_get_max_threads(tpool);
+  }
+  unsigned WorkerPool::numThreads(void) const {
+    return g_thread_pool_get_num_threads(tpool);
+  }
+  unsigned WorkerPool::unprocessed(void) const {
+    return g_thread_pool_unprocessed(tpool);
+  }
+    
+}
+
diff --exclude-from=diff-excludes -urN pan2.svn-orig/pan/tasks/worker-pool.h pan2.svn-decoder-threads-works!/pan/tasks/worker-pool.h
--- pan2.svn-orig/pan/tasks/worker-pool.h	1969-12-31 19:00:00.000000000 -0500
+++ pan2.svn-decoder-threads-works!/pan/tasks/worker-pool.h	2007-03-21 19:18:05.000000000 -0400
@@ -0,0 +1,110 @@
+/* -*- Mode: C++; tab-width: 2; indent-tabs-mode: nil; c-basic-offset: 2 -*- */
+/*
+ * Pan - A Newsreader for Gtk+
+ * Copyright (C) 2007  Calin Culianu <calin@ajvar.org>
+ *
+ * This program is free software; you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation; version 2 of the License.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program; if not, write to the Free Software
+ * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
+ */
+#ifndef worker_pool_H
+#define worker_pool_H
+
+#include <glib.h>
+#include <set>
+
+namespace pan
+{
+  /**
+   * This class basically encapsulates creation and management
+   * of the pan-specific worker-pool mechanism.  The task queue in pan
+   * makes use of this class. */
+  class WorkerPool
+  {
+  public:
+    WorkerPool(int num_threads = -1, bool exclusive = true); ///< creates a pool of num_threads, -1 means no limit, if exclusive then the threads are always running and not shared with other pools (as in glib documentation)
+    ~WorkerPool(); ///< deletes all the threads, may or may not stop workers
+    
+    /// call this on app exit to try and notify all workers to stop!
+    static void cancelAllWorkers(); 
+
+    /** set the limit on number of threads -- semantics just like the glib 
+        funtion */
+    void setMaxThreads(int); 
+    int maxThreads(void) const;
+
+    /// returns number of threads currently runing
+    unsigned numThreads(void) const; 
+    /// returns the number of tasks still unprocess in pool
+    unsigned unprocessed(void) const; 
+
+    struct Listener {
+      virtual ~Listener() {};
+      /** Called in the context of the main thread after enqueued work is done.
+          `data' points to the data passed in when the work was enqueued. */
+      virtual void on_work_complete(void * data) = 0;
+      virtual void on_work_cancelled(void *data) {}
+    };
+
+    class Worker {
+    public:
+      Worker();
+      virtual ~Worker();
+
+      /** Notify of stop request -- called by stopAllWorkers(), 
+          Re-implement if you think you can force a better stop in your 
+          do_work function, otherwise it sets the flag please_stop,
+          which a particular implementation of this class may or may
+          not listen to in order to abort work early. */
+      virtual void cancel() { please_stop = true; }
+      virtual bool was_cancelled() { return please_stop; }
+
+    protected:
+      /** Re-implement in your classes to actually do the work.  Called,
+          by WorkerPool framework in a worker thread. */
+      virtual void do_work(void * data) = 0;
+
+      /** workers implementing do_work() should check this flag and if flag 
+          set, worker should try and stop what it was doing */
+      volatile bool please_stop; 
+      friend class WorkerPool;
+    };
+
+    /** Calls w->do_work(data) for you in the worker thread.
+        When the work is completed, the optional listener is notified
+        *from the main thread*.  Optionally you can tell this function
+        to delete the Worker when the work is done as well.  If deleting,
+        the delete is done after the listener is notified in the main thread.*/
+    void push_work(Worker *w, void * data = 0, Listener * = 0, bool delete_worker_on_completion = false);
+
+
+    typedef std::set<Worker *> WorkerSet;
+
+
+  private:
+    struct Work; 
+
+    static void thr_wrapper(gpointer data, gpointer user_data);
+    static gboolean finalize(gpointer data); /// called in main thread
+    void doWork(Work *); /**< Called in the context of the 
+                            worker thread to do work, deletes work when done */
+
+    mutable GThreadPool *tpool;
+    WorkerSet my_workers;
+
+    /// guarded access to all_workers set -- may initialize heap-allocated static data
+    static WorkerSet & all_workers();  
+  };
+
+};
+
+#endif
